22501146218

Committed 27 Feb 2026 07:42PM UTC coverage: 89.038% (+0.1%) from 88.915%

Build # 22501146218

Build Type

push

github

Committed by

web-flow

Commit Message

Redis cluster sentinel (#968)

* add support for sentinel and clustered redis

Signed-off-by: Gavin Jaeger-Freeborn <gavinfreeborn@gmail.com>

* apply linting

Signed-off-by: Gavin Jaeger-Freeborn <gavinfreeborn@gmail.com>

* added missing redis_utils.py

Signed-off-by: Gavin Jaeger-Freeborn <gavinfreeborn@gmail.com>

* Refactor Redis configuration handling for improved clarity and compatibility

- Update environment variable parsing to require REDIS_HOST in host:port format, enhancing clarity and reducing legacy issues.
- Remove the deprecated REDIS_PORT variable; consolidate port information within REDIS_HOST for consistency.
- Streamline validation and normalization of Redis settings, ensuring errors are raised early and precisely.
- Adjust documentation and comments to reflect new configuration requirements and usage patterns.
- Update tests to cover new behaviors and ensure backward compatibility.

Signed-off-by: Gavin Jaeger-Freeborn <gavinfreeborn@gmail.com>

* Add normalization for Redis configuration at import time

Signed-off-by: Gavin Jaeger-Freeborn <gavinfreeborn@gmail.com>

* Refactor `socketio.py`: Remove duplicate import of `Sentinel`

Signed-off-by: Gavin Jaeger-Freeborn <gavinfreeborn@gmail.com>

* Format with black

Signed-off-by: Gavin Jaeger-Freeborn <gavinfreeborn@gmail.com>

* Fix broken cleanup tests

Signed-off-by: Gavin Jaeger-Freeborn <gavinfreeborn@gmail.com>

* Add tests for Redis utility functions and socket manager

This commit introduces unit tests for Redis utility functions and
enhances error handling in the socket manager related to Sentinel
connectivity. These changes improve the stability and reliability of
the Redis implementation used in the application.

Signed-off-by: Gavin Jaeger-Freeborn <gavinfreeborn@gmail.com>

* Refactor Redis tests for password handling and sentinel host parsing

- Updated tests to reflect that REDIS_PASSWORD is used for both sentinel and master authentication, simplifying configurations... (continued)

Coverage Stats

296 of 335 new or added lines in 11 files covered. (88.36%)

3 existing lines in 2 files now uncovered.

2461 of 2764 relevant lines covered (89.04%)

0.89 hits per line

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

88.68

/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 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."""
    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 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	validate_redis_config()	1✔
149
150	await init_db()	1✔
151	await init_provider(await get_db())	1✔
152
153	# Check Redis availability if adapter is enabled.
154	# Redis is required for both Socket.IO cross-pod messaging AND PyOP token
155	# storage — if it's unreachable the entire auth flow is broken, so we fail
156	# fast here rather than starting and crashing on every request.
157	mode = settings.REDIS_MODE.lower()	1✔
158	if mode == "none":	1✔
159	logger.debug("Redis adapter disabled (REDIS_MODE=none)")	1✔
160	else:
161	try:	1✔
162	reachable = False	1✔
163	if mode == "single":	1✔
164	reachable = can_we_reach_redis(build_redis_url())	1✔
NEW 165	elif mode == "sentinel":	×
NEW 166	hosts = parse_host_port_pairs(settings.REDIS_HOST)	×
NEW 167	reachable = can_we_reach_sentinel(	×
168	hosts, settings.REDIS_SENTINEL_MASTER_NAME
169	)
NEW 170	elif mode == "cluster":	×
NEW 171	hosts = parse_host_port_pairs(settings.REDIS_HOST)	×
NEW 172	reachable = can_we_reach_cluster(hosts)	×
173
174	if reachable:	1✔
175	logger.info(f"Redis adapter is available and ready (mode={mode})")	1✔
176	else:
177	raise RuntimeError(	1✔
178	f"REDIS_MODE={mode} is configured but Redis is not reachable "
179	f"(REDIS_HOST={settings.REDIS_HOST}). "
180	"Ensure Redis is running and accessible, "
181	"or set REDIS_MODE=none to disable Redis."
182	)
183	except RuntimeError:	1✔
184	raise	1✔
UNCOV 185	except Exception as e:	×
UNCOV 186	error_type = _handle_redis_failure("startup Redis check", e)	×
NEW 187	raise RuntimeError(	×
188	f"Redis startup check failed (REDIS_MODE={mode}, "
189	f"REDIS_HOST={settings.REDIS_HOST}, error_type={error_type}): {e}"
190	) from e
191
192	# Robust Webhook Registration
193	if settings.ACAPY_TENANCY == "multi":	1✔
194	logger.debug(	1✔
195	"Starting up in Multi-Tenant Admin Mode",
196	mode="multi",
197	expected_id="Wallet ID (ACAPY_TENANT_WALLET_ID or MT_ACAPY_WALLET_ID)",
198	expected_key="Wallet Key (ACAPY_TENANT_WALLET_KEY or MT_ACAPY_WALLET_KEY)",
199	webhook_registration="Admin API (/multitenancy/wallet/{id})",
200	)
201
202	token_fetcher = None	1✔
203	if settings.ACAPY_TENANT_WALLET_KEY:	1✔
204	token_fetcher = MultiTenantAcapy().get_wallet_token	1✔
205
206	await register_tenant_webhook(	1✔
207	wallet_id=settings.ACAPY_TENANT_WALLET_ID,
208	webhook_url=settings.CONTROLLER_WEB_HOOK_URL,
209	admin_url=settings.ACAPY_ADMIN_URL,
210	api_key=settings.CONTROLLER_API_KEY,
211	admin_api_key=settings.ST_ACAPY_ADMIN_API_KEY,
212	admin_api_key_name=settings.ST_ACAPY_ADMIN_API_KEY_NAME,
213	token_fetcher=token_fetcher,
214	use_admin_api=True,
215	)
216
217	elif settings.ACAPY_TENANCY == "traction":	1✔
218	logger.debug(	1✔
219	"Starting up in Traction Mode",
220	mode="traction",
221	expected_id="Traction Tenant ID (ACAPY_TENANT_WALLET_ID)",
222	expected_key="Traction Tenant API Key (ACAPY_TENANT_WALLET_KEY)",
223	webhook_registration="Tenant API (/tenant/wallet)",
224	)
225
226	token_fetcher = TractionTenantAcapy().get_wallet_token	1✔
227
228	await register_tenant_webhook(	1✔
229	wallet_id=settings.ACAPY_TENANT_WALLET_ID, # Optional/Unused for traction mode registration
230	webhook_url=settings.CONTROLLER_WEB_HOOK_URL,
231	admin_url=settings.ACAPY_ADMIN_URL,
232	api_key=settings.CONTROLLER_API_KEY,
233	admin_api_key=None, # Not used in direct tenant update
234	admin_api_key_name=None,
235	token_fetcher=token_fetcher,
236	use_admin_api=False,
237	)
238
239	logger.info(">>> Starting up app new ...")	1✔
240
241
242	@app.on_event("shutdown")	1✔
243	async def on_tenant_shutdown():	1✔
244	"""Gracefully shutdown services."""
245	logger.info(">>> Shutting down app ...")	×
246
247
248	@app.get("/", tags=["liveness", "readiness"])	1✔
249	@app.get("/health", tags=["liveness", "readiness"])	1✔
250	def main():	1✔
251	return {"status": "ok", "health": "ok"}	1✔
252
253
254	if __name__ == "__main__":	1✔
255	logger.info("main.")	×
256	uvicorn.run(app, host="0.0.0.0", port=5100)	×

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

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