23067145163

Committed 13 Mar 2026 07:27PM UTC coverage: 90.101% (-0.5%) from 90.632%

Build # 23067145163

Build Type

Pull #988

github

Committed by

web-flow

Commit Message

Merge 3776e399b into ac00492c3

Pull Request Pull Request #988: Refactor WebSocket implementation to Server-Sent Events (SSE) for real-time notifications

Coverage Stats

184 of 245 new or added lines in 6 files covered. (75.1%)

8 existing lines in 2 files now uncovered.

2403 of 2667 relevant lines covered (90.1%)

0.9 hits per line

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

39.53

/oidc-controller/api/routers/presentation_request.py

import structlog
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from jinja2 import Template
from pymongo.database import Database

from ..authSessions.crud import AuthSessionCRUD
from ..authSessions.models import AuthSession, AuthSessionState
from ..core.config import settings
from ..routers.sse import notify
from ..routers.oidc import gen_deep_link
from ..core.siem_audit import audit_qr_scanned
from ..db.session import get_db

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

router = APIRouter()


async def toggle_pending(db, auth_session: AuthSession):
    # We need to set this to pending now
    auth_session.proof_status = AuthSessionState.PENDING
    await AuthSessionCRUD(db).patch(auth_session.id, auth_session)
    await notify(str(auth_session.id), "pending")


@router.get("/url/pres_exch/{pres_exch_id}")
async def send_connectionless_proof_req(
    pres_exch_id: str, req: Request, db: Database = Depends(get_db)
):
    """
    If the user scanes the QR code with a mobile camera,
    they will be redirected to a help page.
    """
    # SIEM Audit: Log QR scan before redirect path split to capture all scans
    auth_session: AuthSession = await AuthSessionCRUD(db).get_by_pres_exch_id(
        pres_exch_id
    )
    client_ip = req.client.host if req.client else None
    user_agent = req.headers.get("user-agent")
    audit_qr_scanned(
        session_id=str(auth_session.id),
        scan_method="qr_code",
        client_ip=client_ip,
        user_agent=user_agent,
    )

    # First prepare the response depending on the redirect url
    if ".html" in settings.CONTROLLER_CAMERA_REDIRECT_URL:
        response = RedirectResponse(settings.CONTROLLER_CAMERA_REDIRECT_URL)
    else:
        template_file = open(
            f"{settings.CONTROLLER_TEMPLATE_DIR}/{settings.CONTROLLER_CAMERA_REDIRECT_URL}.html",
            "r",
        ).read()

        wallet_deep_link = gen_deep_link(auth_session)
        template = Template(template_file)

        # If the qrcode was scanned by mobile phone camera toggle the pending flag
        await toggle_pending(db, auth_session)

        response = HTMLResponse(template.render({"wallet_deep_link": wallet_deep_link}))

    if "text/html" in req.headers.get("accept"):
        logger.info("Redirecting to instructions page")
        return response

    auth_session: AuthSession = await AuthSessionCRUD(db).get_by_pres_exch_id(
        pres_exch_id
    )

    # If the qrcode has been scanned, toggle the pending flag
    if auth_session.proof_status is AuthSessionState.NOT_STARTED:
        await toggle_pending(db, auth_session)

        # SIEM Audit: Log QR scan via wallet deep link
        client_ip = req.client.host if req.client else None
        user_agent = req.headers.get("user-agent")
        audit_qr_scanned(
            session_id=str(auth_session.id),
            scan_method="deep_link",
            client_ip=client_ip,
            user_agent=user_agent,
        )

    msg = auth_session.presentation_request_msg

    logger.debug(msg)
    return JSONResponse(msg)

1	import structlog	1✔
2	from fastapi import APIRouter, Depends, Request	1✔
3	from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse	1✔
4	from jinja2 import Template	1✔
5	from pymongo.database import Database	1✔
6
7	from ..authSessions.crud import AuthSessionCRUD	1✔
8	from ..authSessions.models import AuthSession, AuthSessionState	1✔
9	from ..core.config import settings	1✔
10	from ..routers.sse import notify	1✔
11	from ..routers.oidc import gen_deep_link	1✔
12	from ..core.siem_audit import audit_qr_scanned	1✔
13	from ..db.session import get_db	1✔
14
15	logger: structlog.typing.FilteringBoundLogger = structlog.getLogger(__name__)	1✔
16
17	router = APIRouter()	1✔
18
19
20	async def toggle_pending(db, auth_session: AuthSession):	1✔
21	# We need to set this to pending now
22	auth_session.proof_status = AuthSessionState.PENDING	×
23	await AuthSessionCRUD(db).patch(auth_session.id, auth_session)	×
NEW 24	await notify(str(auth_session.id), "pending")	×
25
26
27	@router.get("/url/pres_exch/{pres_exch_id}")	1✔
28	async def send_connectionless_proof_req(	1✔
29	pres_exch_id: str, req: Request, db: Database = Depends(get_db)
30	):
31	"""
32	If the user scanes the QR code with a mobile camera,
33	they will be redirected to a help page.
34	"""
35	# SIEM Audit: Log QR scan before redirect path split to capture all scans
36	auth_session: AuthSession = await AuthSessionCRUD(db).get_by_pres_exch_id(	×
37	pres_exch_id
38	)
39	client_ip = req.client.host if req.client else None	×
40	user_agent = req.headers.get("user-agent")	×
41	audit_qr_scanned(	×
42	session_id=str(auth_session.id),
43	scan_method="qr_code",
44	client_ip=client_ip,
45	user_agent=user_agent,
46	)
47
48	# First prepare the response depending on the redirect url
49	if ".html" in settings.CONTROLLER_CAMERA_REDIRECT_URL:	×
50	response = RedirectResponse(settings.CONTROLLER_CAMERA_REDIRECT_URL)	×
51	else:
52	template_file = open(	×
53	f"{settings.CONTROLLER_TEMPLATE_DIR}/{settings.CONTROLLER_CAMERA_REDIRECT_URL}.html",
54	"r",
55	).read()
56
57	wallet_deep_link = gen_deep_link(auth_session)	×
58	template = Template(template_file)	×
59
60	# If the qrcode was scanned by mobile phone camera toggle the pending flag
61	await toggle_pending(db, auth_session)	×
62
63	response = HTMLResponse(template.render({"wallet_deep_link": wallet_deep_link}))	×
64
65	if "text/html" in req.headers.get("accept"):	×
66	logger.info("Redirecting to instructions page")	×
67	return response	×
68
69	auth_session: AuthSession = await AuthSessionCRUD(db).get_by_pres_exch_id(	×
70	pres_exch_id
71	)
72
73	# If the qrcode has been scanned, toggle the pending flag
74	if auth_session.proof_status is AuthSessionState.NOT_STARTED:	×
75	await toggle_pending(db, auth_session)	×
76
77	# SIEM Audit: Log QR scan via wallet deep link
78	client_ip = req.client.host if req.client else None	×
79	user_agent = req.headers.get("user-agent")	×
80	audit_qr_scanned(	×
81	session_id=str(auth_session.id),
82	scan_method="deep_link",
83	client_ip=client_ip,
84	user_agent=user_agent,
85	)
86
87	msg = auth_session.presentation_request_msg	×
88
89	logger.debug(msg)	×
90	return JSONResponse(msg)	×

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

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