22330627754

Committed 24 Feb 2026 12:12AM UTC coverage: 89.109% (+0.2%) from 88.915%

Build # 22330627754

Build Type

Pull #968

github

Committed by

web-flow

Commit Message

Merge 5bfdab059 into e0695bac9

Pull Request Pull Request #968: Redis cluster sentinel

Run Details

308 of 347 new or added lines in 11 files covered. (88.76%)

3 existing lines in 2 files now uncovered.

2479 of 2782 relevant lines covered (89.11%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.79

/oidc-controller/api/main.py

import traceback
import structlog
import os
import time
import uuid
from pathlib import Path

import uvicorn
from api.core.config import settings
from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi import status as http_status
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles

from .db.session import get_db, init_db
from .routers import (
    acapy_handler,
    cleanup,
    oidc,
    presentation_request,
    well_known_oid_config,
)
from .verificationConfigs.router import router as ver_configs_router
from .clientConfigurations.router import router as client_config_router
from .routers.socketio import (
    sio_app,
    _handle_redis_failure,
    can_we_reach_redis,
    can_we_reach_cluster,
    can_we_reach_sentinel,
)
from .core.redis_utils import parse_host_port_pairs, build_redis_url
from api.core.oidc.provider import init_provider
from api.core.webhook_utils import register_tenant_webhook
from api.core.acapy.config import MultiTenantAcapy, TractionTenantAcapy
from api.core.config import normalize_redis_config, validate_redis_config

logger: structlog.typing.FilteringBoundLogger = structlog.getLogger(__name__)

# Setup loggers
logging_file_path = os.environ.get(
    "LOG_CONFIG_PATH", (Path(__file__).parent / "logging.conf").resolve()
)


os.environ["TZ"] = settings.TIMEZONE
time.tzset()


def get_application() -> FastAPI:
    application = FastAPI(
        title=settings.TITLE,
        description=settings.DESCRIPTION,
        debug=settings.DEBUG,
        # middleware=None,
    )
    return application


app = get_application()

# Serve static assets for the frontend
app.mount(
    "/static",
    StaticFiles(directory=(settings.CONTROLLER_TEMPLATE_DIR + "/assets")),
    name="static",
)

# Include routers
app.include_router(ver_configs_router, prefix="/ver_configs", tags=["ver_configs"])
app.include_router(client_config_router, prefix="/clients", tags=["oidc_clients"])
app.include_router(well_known_oid_config.router, tags=[".well-known"])
app.include_router(
    oidc.router, tags=["OpenID Connect Provider"], include_in_schema=False
)
app.include_router(acapy_handler.router, prefix="/webhooks", include_in_schema=False)
app.include_router(presentation_request.router, include_in_schema=False)
app.include_router(cleanup.router, tags=["cleanup"])

# DEPRECATED PATHS - For backwards compatibility with vc-authn-oidc 1.0
app.include_router(
    oidc.router, prefix="/vc/connect", tags=["oidc-deprecated"], include_in_schema=False
)

# Connect the websocket server to run within the FastAPI app
app.mount("/ws", sio_app)

origins = ["*"]

if origins:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )


@app.middleware("http")
async def logging_middleware(request: Request, call_next) -> Response:
    structlog.threadlocal.clear_threadlocal()
    structlog.threadlocal.bind_threadlocal(
        logger="uvicorn.access",
        request_id=str(uuid.uuid4()),
        cookies=request.cookies,
        scope=request.scope,
        url=str(request.url),
    )
    start_time = time.time()
    try:
        response: Response = await call_next(request)
        return response
    except Exception:
        process_time = time.time() - start_time
        logger.info(
            "failed to process a request",
            status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
            process_time=process_time,
        )

        # Need to explicitly log the traceback
        logger.error(traceback.format_exc())

        return JSONResponse(
            status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={
                "status": "error",
                "message": "Internal Server Error",
                "process_time": process_time,
            },
        )
    else:
        process_time = time.time() - start_time
        logger.info(
            "processed a request",
            status_code=response.status_code,
            process_time=process_time,
        )


@app.on_event("startup")
async def on_tenant_startup():
    """Register any events we need to respond to."""
    # Normalize then validate Redis configuration early
    normalize_redis_config()
    validate_redis_config()

    await init_db()
    await init_provider(await get_db())

    # Check Redis availability if adapter is enabled.
    # Redis is required for both Socket.IO cross-pod messaging AND PyOP token
    # storage — if it's unreachable the entire auth flow is broken, so we fail
    # fast here rather than starting and crashing on every request.
    mode = settings.REDIS_MODE.lower()
    if mode == "none":
        logger.debug("Redis adapter disabled (REDIS_MODE=none)")
    else:
        try:
            reachable = False
            if mode == "single":
                reachable = can_we_reach_redis(build_redis_url())
            elif mode == "sentinel":
                hosts = parse_host_port_pairs(settings.REDIS_HOST)
                reachable = can_we_reach_sentinel(
                    hosts, settings.REDIS_SENTINEL_MASTER_NAME
                )
            elif mode == "cluster":
                hosts = parse_host_port_pairs(settings.REDIS_HOST)
                reachable = can_we_reach_cluster(hosts)

            if reachable:
                logger.info(f"Redis adapter is available and ready (mode={mode})")
            else:
                raise RuntimeError(
                    f"REDIS_MODE={mode} is configured but Redis is not reachable "
                    f"(REDIS_HOST={settings.REDIS_HOST}). "
                    "Ensure Redis is running and accessible, "
                    "or set REDIS_MODE=none to disable Redis."
                )
        except RuntimeError:
            raise
        except Exception as e:
            error_type = _handle_redis_failure("startup Redis check", e)
            raise RuntimeError(
                f"Redis startup check failed (REDIS_MODE={mode}, "
                f"REDIS_HOST={settings.REDIS_HOST}, error_type={error_type}): {e}"
            ) from e

    # Robust Webhook Registration
    if settings.ACAPY_TENANCY == "multi":
        logger.debug(
            "Starting up in Multi-Tenant Admin Mode",
            mode="multi",
            expected_id="Wallet ID (ACAPY_TENANT_WALLET_ID or MT_ACAPY_WALLET_ID)",
            expected_key="Wallet Key (ACAPY_TENANT_WALLET_KEY or MT_ACAPY_WALLET_KEY)",
            webhook_registration="Admin API (/multitenancy/wallet/{id})",
        )

        token_fetcher = None
        if settings.ACAPY_TENANT_WALLET_KEY:
            token_fetcher = MultiTenantAcapy().get_wallet_token

        await register_tenant_webhook(
            wallet_id=settings.ACAPY_TENANT_WALLET_ID,
            webhook_url=settings.CONTROLLER_WEB_HOOK_URL,
            admin_url=settings.ACAPY_ADMIN_URL,
            api_key=settings.CONTROLLER_API_KEY,
            admin_api_key=settings.ST_ACAPY_ADMIN_API_KEY,
            admin_api_key_name=settings.ST_ACAPY_ADMIN_API_KEY_NAME,
            token_fetcher=token_fetcher,
            use_admin_api=True,
        )

    elif settings.ACAPY_TENANCY == "traction":
        logger.debug(
            "Starting up in Traction Mode",
            mode="traction",
            expected_id="Traction Tenant ID (ACAPY_TENANT_WALLET_ID)",
            expected_key="Traction Tenant API Key (ACAPY_TENANT_WALLET_KEY)",
            webhook_registration="Tenant API (/tenant/wallet)",
        )

        token_fetcher = TractionTenantAcapy().get_wallet_token

        await register_tenant_webhook(
            wallet_id=settings.ACAPY_TENANT_WALLET_ID,  # Optional/Unused for traction mode registration
            webhook_url=settings.CONTROLLER_WEB_HOOK_URL,
            admin_url=settings.ACAPY_ADMIN_URL,
            api_key=settings.CONTROLLER_API_KEY,
            admin_api_key=None,  # Not used in direct tenant update
            admin_api_key_name=None,
            token_fetcher=token_fetcher,
            use_admin_api=False,
        )

    logger.info(">>> Starting up app new ...")


@app.on_event("shutdown")
async def on_tenant_shutdown():
    """Gracefully shutdown services."""
    logger.info(">>> Shutting down app ...")


@app.get("/", tags=["liveness", "readiness"])
@app.get("/health", tags=["liveness", "readiness"])
def main():
    return {"status": "ok", "health": "ok"}


if __name__ == "__main__":
    logger.info("main.")
    uvicorn.run(app, host="0.0.0.0", port=5100)

1	import traceback	1✔
2	import structlog	1✔
3	import os	1✔
4	import time	1✔
5	import uuid	1✔
6	from pathlib import Path	1✔
7
8	import uvicorn	1✔
9	from api.core.config import settings	1✔
10	from fastapi import FastAPI	1✔
11	from starlette.requests import Request	1✔
12	from starlette.responses import Response	1✔
13	from fastapi.middleware.cors import CORSMiddleware	1✔
14	from fastapi import status as http_status	1✔
15	from fastapi.responses import JSONResponse	1✔
16	from fastapi.staticfiles import StaticFiles	1✔
17
18	from .db.session import get_db, init_db	1✔
19	from .routers import (	1✔
20	acapy_handler,
21	cleanup,
22	oidc,
23	presentation_request,
24	well_known_oid_config,
25	)
26	from .verificationConfigs.router import router as ver_configs_router	1✔
27	from .clientConfigurations.router import router as client_config_router	1✔
28	from .routers.socketio import (	1✔
29	sio_app,
30	_handle_redis_failure,
31	can_we_reach_redis,
32	can_we_reach_cluster,
33	can_we_reach_sentinel,
34	)
35	from .core.redis_utils import parse_host_port_pairs, build_redis_url	1✔
36	from api.core.oidc.provider import init_provider	1✔
37	from api.core.webhook_utils import register_tenant_webhook	1✔
38	from api.core.acapy.config import MultiTenantAcapy, TractionTenantAcapy	1✔
39	from api.core.config import normalize_redis_config, validate_redis_config	1✔
40
41	logger: structlog.typing.FilteringBoundLogger = structlog.getLogger(__name__)	1✔
42
43	# Setup loggers
44	logging_file_path = os.environ.get(	1✔
45	"LOG_CONFIG_PATH", (Path(__file__).parent / "logging.conf").resolve()
46	)
47
48
49	os.environ["TZ"] = settings.TIMEZONE	1✔
50	time.tzset()	1✔
51
52
53	def get_application() -> FastAPI:	1✔
54	application = FastAPI(	1✔
55	title=settings.TITLE,
56	description=settings.DESCRIPTION,
57	debug=settings.DEBUG,
58	# middleware=None,
59	)
60	return application	1✔
61
62
63	app = get_application()	1✔
64
65	# Serve static assets for the frontend
66	app.mount(	1✔
67	"/static",
68	StaticFiles(directory=(settings.CONTROLLER_TEMPLATE_DIR + "/assets")),
69	name="static",
70	)
71
72	# Include routers
73	app.include_router(ver_configs_router, prefix="/ver_configs", tags=["ver_configs"])	1✔
74	app.include_router(client_config_router, prefix="/clients", tags=["oidc_clients"])	1✔
75	app.include_router(well_known_oid_config.router, tags=[".well-known"])	1✔
76	app.include_router(	1✔
77	oidc.router, tags=["OpenID Connect Provider"], include_in_schema=False
78	)
79	app.include_router(acapy_handler.router, prefix="/webhooks", include_in_schema=False)	1✔
80	app.include_router(presentation_request.router, include_in_schema=False)	1✔
81	app.include_router(cleanup.router, tags=["cleanup"])	1✔
82
83	# DEPRECATED PATHS - For backwards compatibility with vc-authn-oidc 1.0
84	app.include_router(	1✔
85	oidc.router, prefix="/vc/connect", tags=["oidc-deprecated"], include_in_schema=False
86	)
87
88	# Connect the websocket server to run within the FastAPI app
89	app.mount("/ws", sio_app)	1✔
90
91	origins = ["*"]	1✔
92
93	if origins:	1✔
94	app.add_middleware(	1✔
95	CORSMiddleware,
96	allow_origins=origins,
97	allow_credentials=True,
98	allow_methods=["*"],
99	allow_headers=["*"],
100	)
101
102
103	@app.middleware("http")	1✔
104	async def logging_middleware(request: Request, call_next) -> Response:	1✔
105	structlog.threadlocal.clear_threadlocal()	1✔
106	structlog.threadlocal.bind_threadlocal(	1✔
107	logger="uvicorn.access",
108	request_id=str(uuid.uuid4()),
109	cookies=request.cookies,
110	scope=request.scope,
111	url=str(request.url),
112	)
113	start_time = time.time()	1✔
114	try:	1✔
115	response: Response = await call_next(request)	1✔
116	return response	1✔
117	except Exception:	1✔
118	process_time = time.time() - start_time	1✔
119	logger.info(	1✔
120	"failed to process a request",
121	status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
122	process_time=process_time,
123	)
124
125	# Need to explicitly log the traceback
126	logger.error(traceback.format_exc())	1✔
127
128	return JSONResponse(	1✔
129	status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
130	content={
131	"status": "error",
132	"message": "Internal Server Error",
133	"process_time": process_time,
134	},
135	)
136	else:
137	process_time = time.time() - start_time
138	logger.info(
139	"processed a request",
140	status_code=response.status_code,
141	process_time=process_time,
142	)
143
144
145	@app.on_event("startup")	1✔
146	async def on_tenant_startup():	1✔
147	"""Register any events we need to respond to."""
148	# Normalize then validate Redis configuration early
149	normalize_redis_config()	1✔
150	validate_redis_config()	1✔
151
152	await init_db()	1✔
153	await init_provider(await get_db())	1✔
154
155	# Check Redis availability if adapter is enabled.
156	# Redis is required for both Socket.IO cross-pod messaging AND PyOP token
157	# storage — if it's unreachable the entire auth flow is broken, so we fail
158	# fast here rather than starting and crashing on every request.
159	mode = settings.REDIS_MODE.lower()	1✔
160	if mode == "none":	1✔
161	logger.debug("Redis adapter disabled (REDIS_MODE=none)")	1✔
162	else:
163	try:	1✔
164	reachable = False	1✔
165	if mode == "single":	1✔
166	reachable = can_we_reach_redis(build_redis_url())	1✔
NEW 167	elif mode == "sentinel":	×
NEW 168	hosts = parse_host_port_pairs(settings.REDIS_HOST)	×
NEW 169	reachable = can_we_reach_sentinel(	×
170	hosts, settings.REDIS_SENTINEL_MASTER_NAME
171	)
NEW 172	elif mode == "cluster":	×
NEW 173	hosts = parse_host_port_pairs(settings.REDIS_HOST)	×
NEW 174	reachable = can_we_reach_cluster(hosts)	×
175
176	if reachable:	1✔
177	logger.info(f"Redis adapter is available and ready (mode={mode})")	1✔
178	else:
179	raise RuntimeError(	1✔
180	f"REDIS_MODE={mode} is configured but Redis is not reachable "
181	f"(REDIS_HOST={settings.REDIS_HOST}). "
182	"Ensure Redis is running and accessible, "
183	"or set REDIS_MODE=none to disable Redis."
184	)
185	except RuntimeError:	1✔
186	raise	1✔
UNCOV 187	except Exception as e:	×
UNCOV 188	error_type = _handle_redis_failure("startup Redis check", e)	×
NEW 189	raise RuntimeError(	×
190	f"Redis startup check failed (REDIS_MODE={mode}, "
191	f"REDIS_HOST={settings.REDIS_HOST}, error_type={error_type}): {e}"
192	) from e
193
194	# Robust Webhook Registration
195	if settings.ACAPY_TENANCY == "multi":	1✔
196	logger.debug(	1✔
197	"Starting up in Multi-Tenant Admin Mode",
198	mode="multi",
199	expected_id="Wallet ID (ACAPY_TENANT_WALLET_ID or MT_ACAPY_WALLET_ID)",
200	expected_key="Wallet Key (ACAPY_TENANT_WALLET_KEY or MT_ACAPY_WALLET_KEY)",
201	webhook_registration="Admin API (/multitenancy/wallet/{id})",
202	)
203
204	token_fetcher = None	1✔
205	if settings.ACAPY_TENANT_WALLET_KEY:	1✔
206	token_fetcher = MultiTenantAcapy().get_wallet_token	1✔
207
208	await register_tenant_webhook(	1✔
209	wallet_id=settings.ACAPY_TENANT_WALLET_ID,
210	webhook_url=settings.CONTROLLER_WEB_HOOK_URL,
211	admin_url=settings.ACAPY_ADMIN_URL,
212	api_key=settings.CONTROLLER_API_KEY,
213	admin_api_key=settings.ST_ACAPY_ADMIN_API_KEY,
214	admin_api_key_name=settings.ST_ACAPY_ADMIN_API_KEY_NAME,
215	token_fetcher=token_fetcher,
216	use_admin_api=True,
217	)
218
219	elif settings.ACAPY_TENANCY == "traction":	1✔
220	logger.debug(	1✔
221	"Starting up in Traction Mode",
222	mode="traction",
223	expected_id="Traction Tenant ID (ACAPY_TENANT_WALLET_ID)",
224	expected_key="Traction Tenant API Key (ACAPY_TENANT_WALLET_KEY)",
225	webhook_registration="Tenant API (/tenant/wallet)",
226	)
227
228	token_fetcher = TractionTenantAcapy().get_wallet_token	1✔
229
230	await register_tenant_webhook(	1✔
231	wallet_id=settings.ACAPY_TENANT_WALLET_ID, # Optional/Unused for traction mode registration
232	webhook_url=settings.CONTROLLER_WEB_HOOK_URL,
233	admin_url=settings.ACAPY_ADMIN_URL,
234	api_key=settings.CONTROLLER_API_KEY,
235	admin_api_key=None, # Not used in direct tenant update
236	admin_api_key_name=None,
237	token_fetcher=token_fetcher,
238	use_admin_api=False,
239	)
240
241	logger.info(">>> Starting up app new ...")	1✔
242
243
244	@app.on_event("shutdown")	1✔
245	async def on_tenant_shutdown():	1✔
246	"""Gracefully shutdown services."""
247	logger.info(">>> Shutting down app ...")	×
248
249
250	@app.get("/", tags=["liveness", "readiness"])	1✔
251	@app.get("/health", tags=["liveness", "readiness"])	1✔
252	def main():	1✔
253	return {"status": "ok", "health": "ok"}	1✔
254
255
256	if __name__ == "__main__":	1✔
257	logger.info("main.")	×
258	uvicorn.run(app, host="0.0.0.0", port=5100)	×

openwallet-foundation / acapy-vc-authn-oidc / 22330627754

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous