22593354327

Committed 02 Mar 2026 08:00PM UTC coverage: 89.577% (+0.5%) from 89.038%

Build # 22593354327

Build Type

Pull #971

github

Committed by

web-flow

Commit Message

Merge 1ada79187 into 21c331d7b

Pull Request Pull Request #971: Replace requests with httpx for async HTTP

Coverage Stats

168 of 178 new or added lines in 9 files covered. (94.38%)

9 existing lines in 2 files now uncovered.

2458 of 2744 relevant lines covered (89.58%)

0.9 hits per line

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

87.27

/oidc-controller/api/main.py

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

import httpx
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 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."""
    app.state.http_client = httpx.AsyncClient(
        timeout=httpx.Timeout(settings.ACAPY_REQUEST_TIMEOUT),
        limits=httpx.Limits(
            max_keepalive_connections=20,
            max_connections=100,
            keepalive_expiry=30,
        ),
    )

    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(app.state.http_client).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,
            http_client=app.state.http_client,
            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(app.state.http_client).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,
            http_client=app.state.http_client,
            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 ...")
    if hasattr(app.state, "http_client"):
        await app.state.http_client.aclose()


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

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

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