22197536649

Committed 19 Feb 2026 07:50PM UTC coverage: 88.915% (+0.1%) from 88.818%

Build # 22197536649

Build Type

Pull #969

github

Committed by

web-flow

Commit Message

Merge bd6e7e2a1 into 405ee33da

Pull Request Pull Request #969: Feat/siam audit events

Run Details

154 of 170 new or added lines in 4 files covered. (90.59%)

4 existing lines in 2 files now uncovered.

2230 of 2508 relevant lines covered (88.92%)

0.89 hits per line

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

36.96

/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 ..core.siam_audit import audit_qr_scanned
from ..db.session import get_db
from ..routers.oidc import gen_deep_link
from ..routers.socketio import get_socket_id_for_pid, safe_emit, sio

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)
    sid = await get_socket_id_for_pid(str(auth_session.id), db)
    if sid:
        await safe_emit("status", {"status": "pending"}, to=sid)


@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.
    """
    # SIAM 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
    )

    # Get the websocket session
    sid = await get_socket_id_for_pid(str(auth_session.id), db)

    # 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)

        # SIAM 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 ..core.siam_audit import audit_qr_scanned	1✔
11	from ..db.session import get_db	1✔
12	from ..routers.oidc import gen_deep_link	1✔
13	from ..routers.socketio import get_socket_id_for_pid, safe_emit, sio	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)	×
24	sid = await get_socket_id_for_pid(str(auth_session.id), db)	×
25	if sid:	×
26	await safe_emit("status", {"status": "pending"}, to=sid)	×
27
28
29	@router.get("/url/pres_exch/{pres_exch_id}")	1✔
30	async def send_connectionless_proof_req(	1✔
31	pres_exch_id: str, req: Request, db: Database = Depends(get_db)
32	):
33	"""
34	If the user scanes the QR code with a mobile camera,
35	they will be redirected to a help page.
36	"""
37	# SIAM Audit: Log QR scan before redirect path split to capture all scans
NEW 38	auth_session: AuthSession = await AuthSessionCRUD(db).get_by_pres_exch_id(	×
39	pres_exch_id
40	)
NEW 41	client_ip = req.client.host if req.client else None	×
NEW 42	user_agent = req.headers.get("user-agent")	×
NEW 43	audit_qr_scanned(	×
44	session_id=str(auth_session.id),
45	scan_method="qr_code",
46	client_ip=client_ip,
47	user_agent=user_agent,
48	)
49
50	# First prepare the response depending on the redirect url
51	if ".html" in settings.CONTROLLER_CAMERA_REDIRECT_URL:	×
52	response = RedirectResponse(settings.CONTROLLER_CAMERA_REDIRECT_URL)	×
53	else:
54	template_file = open(	×
55	f"{settings.CONTROLLER_TEMPLATE_DIR}/{settings.CONTROLLER_CAMERA_REDIRECT_URL}.html",
56	"r",
57	).read()
58
59	wallet_deep_link = gen_deep_link(auth_session)	×
UNCOV 60	template = Template(template_file)	×
61
62	# If the qrcode was scanned by mobile phone camera toggle the pending flag
63	await toggle_pending(db, auth_session)	×
64
UNCOV 65	response = HTMLResponse(template.render({"wallet_deep_link": wallet_deep_link}))	×
66
67	if "text/html" in req.headers.get("accept"):	×
68	logger.info("Redirecting to instructions page")	×
69	return response	×
70
71	auth_session: AuthSession = await AuthSessionCRUD(db).get_by_pres_exch_id(	×
72	pres_exch_id
73	)
74
75	# Get the websocket session
76	sid = await get_socket_id_for_pid(str(auth_session.id), db)	×
77
78	# If the qrcode has been scanned, toggle the pending flag
79	if auth_session.proof_status is AuthSessionState.NOT_STARTED:	×
80	await toggle_pending(db, auth_session)	×
81
82	# SIAM Audit: Log QR scan via wallet deep link
NEW 83	client_ip = req.client.host if req.client else None	×
NEW 84	user_agent = req.headers.get("user-agent")	×
NEW 85	audit_qr_scanned(	×
86	session_id=str(auth_session.id),
87	scan_method="deep_link",
88	client_ip=client_ip,
89	user_agent=user_agent,
90	)
91
UNCOV 92	msg = auth_session.presentation_request_msg	×
93
94	logger.debug(msg)	×
95	return JSONResponse(msg)	×

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

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