19545712621

Committed 20 Nov 2025 05:30PM UTC coverage: 92.267% (-0.1%) from 92.384%

Build # 19545712621

Build Type

Pull #906

github

Committed by

web-flow

Commit Message

Merge 26080a16b into 4906a7603

Pull Request Pull Request #906: Feat/manage script testing

Run Details

12 of 15 new or added lines in 3 files covered. (80.0%)

30 existing lines in 2 files now uncovered.

1587 of 1720 relevant lines covered (92.27%)

0.92 hits per line

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

96.34

/oidc-controller/api/core/oidc/issue_token_service.py

import canonicaljson
import dataclasses
import json
import hashlib
from datetime import datetime
from typing import Any

import structlog
from oic.oic.message import OpenIDSchema
from pydantic import BaseModel

from ...authSessions.models import AuthSession
from ...verificationConfigs.models import ReqAttr, VerificationConfig
from ...core.models import RevealedAttribute
from ..config import settings

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

PROOF_CLAIMS_ATTRIBUTE_NAME = "vc_presented_attributes"


@dataclasses.dataclass(frozen=True)
class Claim(BaseModel):
    type: str
    value: str


class Token(BaseModel):
    creation_time: datetime = datetime.now()
    issuer: str
    audiences: list[str]
    lifetime: int
    claims: dict[str, Any]

    @classmethod
    def get_claims(
        cls, auth_session: AuthSession, ver_config: VerificationConfig
    ) -> dict[str, str]:
        """Converts vc presentation values to oidc claims"""
        oidc_claims: list[Claim] = [
            Claim(
                type="pres_req_conf_id",
                value=auth_session.request_parameters["pres_req_conf_id"],
            ),
            Claim(type="acr", value="vc_authn"),
        ]
        # subject claim

        oidc_claims.append(
            Claim(type="nonce", value=auth_session.request_parameters["nonce"])
        )

        presentation_claims: dict[str, Claim] = {}

        # Handle in-flight sessions during configuration changes (e.g., switching 'indy' to 'anoncreds').
        # If the current config setting doesn't match the stored record,
        # fallback to the format actually found in the database.
        pres_request = auth_session.presentation_exchange.get("pres_request", {})
        proof_format = settings.ACAPY_PROOF_FORMAT

        if proof_format not in pres_request:
            if "indy" in pres_request:
                logger.debug(
                    f"Configured proof format '{proof_format}' not found in record, falling back to 'indy'"
                )
                proof_format = "indy"
            elif "anoncreds" in pres_request:
                logger.debug(
                    f"Configured proof format '{proof_format}' not found in record, falling back to 'anoncreds'"
                )
                proof_format = "anoncreds"
            # If neither found, we stick with settings.ACAPY_PROOF_FORMAT and let it likely fail below

        logger.info(
            "Extracted requested attributes from presentation request",
            requested_attributes=auth_session.presentation_exchange["pres_request"][
                proof_format
            ]["requested_attributes"],
        )

        referent: str
        requested_attr: ReqAttr
        try:
            for referent, requested_attrdict in auth_session.presentation_exchange[
                "pres_request"
            ][proof_format]["requested_attributes"].items():
                requested_attr = ReqAttr(**requested_attrdict)
                logger.debug(
                    f"Processing referent: {referent}, requested_attr: {requested_attr}"
                )
                revealed_attrs: dict[str, RevealedAttribute] = (
                    auth_session.presentation_exchange["pres"][proof_format][
                        "requested_proof"
                    ]["revealed_attr_groups"]
                )
                logger.debug(f"revealed_attrs: {revealed_attrs}")
                # loop through each value and put it in token as a claim
                for attr_name in requested_attr.names:
                    logger.debug(f"AttrName: {attr_name}")
                    presentation_claims[attr_name] = Claim(
                        type=attr_name,
                        value=revealed_attrs[referent]["values"][attr_name]["raw"],
                    )
                    logger.debug(f"Compiled presentation_claims: {presentation_claims}")
        except Exception as err:
            logger.error(
                f"An exception occurred while extracting the proof claims: {err}"
            )
            raise RuntimeError(err)

        proof_claims = json.dumps(
            {c.type: c.value for c in presentation_claims.values()}
        )
        # look at all presentation_claims for one
        # matching the configured subject_identifier, if any
        sub_id_claim = presentation_claims.get(ver_config.subject_identifier)

        pres_req_conf_id_suffix = (
            f"@{auth_session.request_parameters['pres_req_conf_id']}"
        )

        if sub_id_claim:
            # add sub and append presentation_claims
            oidc_claims.append(
                Claim(
                    type="sub",
                    value=sub_id_claim.value + pres_req_conf_id_suffix,
                )
            )

        elif ver_config.generate_consistent_identifier:
            # Do not create a sub based on the proof claims if the
            # user requests a generated identifier
            # Generate a SHA256 hash of the canonicaljson encoded proof_claims
            encoded_json: bytes = canonicaljson.encode_canonical_json(proof_claims)
            sha256_hash = hashlib.sha256(
                encoded_json + pres_req_conf_id_suffix.encode()
            ).hexdigest()
            oidc_claims.append(
                Claim(
                    type="sub",
                    value=sha256_hash,
                )
            )

        result = {c.type: c.value for c in oidc_claims}
        result[PROOF_CLAIMS_ATTRIBUTE_NAME] = proof_claims

        # TODO: Remove after full transistion to v2.0
        # Add the presentation claims to the result as keys
        # for backwards compatibility [v1.0]
        if ver_config.include_v1_attributes:
            for key, value in presentation_claims.items():
                result[key] = value.value

        return result

    # TODO: Determine if this is useful to keep, and remove it if it's not.
    # It is currently unused.
    # renames and calculates dict members appropriate to
    # https://openid.net/specs/openid-connect-core-1_0.html#IDToken
    # and
    # https://github.com/OpenIDC/pyoidc/blob/26ea5121239dad03c5c5551cca149cb984df1ec9/src/oic/oic/message.py#L720
    def idtoken_dict(self, nonce: str) -> dict:
        """Converts oidc claims to IdToken attribute names"""

        result = {}  # nest VC attribute claims under the key=pres_req_conf_id

        # for type, value in self.claims.items():
        #     result[type] = value

        result["exp"] = int(round(datetime.now().timestamp())) + self.lifetime
        result["aud"] = self.audiences
        result["nonce"] = nonce

        result.update(self.claims)

        # identify if any standardclaims were provided in the proof and return
        # them at the top level.
        # https://openid.net/specs/openid-connect-core-1_0.html#StandardClaims

        # make copy of dict
        r2 = result.copy()
        # add nested values to top level
        r2.update(json.loads(self.claims[PROOF_CLAIMS_ATTRIBUTE_NAME]))
        # only keep ones that match the OpenIDschema
        r2 = {
            key: r2[key]
            for key in set(r2.keys()).intersection(set(OpenIDSchema().c_param.keys()))
        }

        # verify with library schema
        standard_claims = OpenIDSchema().from_dict(r2)
        standard_claims.verify()
        # add to the top level of the dict.
        for key, value in standard_claims.to_dict().items():
            result[key] = value

        return result

1	import canonicaljson	1✔
2	import dataclasses	1✔
3	import json	1✔
4	import hashlib	1✔
5	from datetime import datetime	1✔
6	from typing import Any	1✔
7
8	import structlog	1✔
9	from oic.oic.message import OpenIDSchema	1✔
10	from pydantic import BaseModel	1✔
11
12	from ...authSessions.models import AuthSession	1✔
13	from ...verificationConfigs.models import ReqAttr, VerificationConfig	1✔
14	from ...core.models import RevealedAttribute	1✔
15	from ..config import settings	1✔
16
17	logger: structlog.typing.FilteringBoundLogger = structlog.getLogger(__name__)	1✔
18
19	PROOF_CLAIMS_ATTRIBUTE_NAME = "vc_presented_attributes"	1✔
20
21
22	@dataclasses.dataclass(frozen=True)	1✔
23	class Claim(BaseModel):	1✔
24	type: str	1✔
25	value: str	1✔
26
27
28	class Token(BaseModel):	1✔
29	creation_time: datetime = datetime.now()	1✔
30	issuer: str	1✔
31	audiences: list[str]	1✔
32	lifetime: int	1✔
33	claims: dict[str, Any]	1✔
34
35	@classmethod	1✔
36	def get_claims(	1✔
37	cls, auth_session: AuthSession, ver_config: VerificationConfig
38	) -> dict[str, str]:
39	"""Converts vc presentation values to oidc claims"""
40	oidc_claims: list[Claim] = [	1✔
41	Claim(
42	type="pres_req_conf_id",
43	value=auth_session.request_parameters["pres_req_conf_id"],
44	),
45	Claim(type="acr", value="vc_authn"),
46	]
47	# subject claim
48
49	oidc_claims.append(	1✔
50	Claim(type="nonce", value=auth_session.request_parameters["nonce"])
51	)
52
53	presentation_claims: dict[str, Claim] = {}	1✔
54
55	# Handle in-flight sessions during configuration changes (e.g., switching 'indy' to 'anoncreds').
56	# If the current config setting doesn't match the stored record,
57	# fallback to the format actually found in the database.
58	pres_request = auth_session.presentation_exchange.get("pres_request", {})	1✔
59	proof_format = settings.ACAPY_PROOF_FORMAT	1✔
60
61	if proof_format not in pres_request:	1✔
62	if "indy" in pres_request:	1✔
63	logger.debug(	1✔
64	f"Configured proof format '{proof_format}' not found in record, falling back to 'indy'"
65	)
66	proof_format = "indy"	1✔
NEW UNCOV 67	elif "anoncreds" in pres_request:	×
NEW UNCOV 68	logger.debug(	×
69	f"Configured proof format '{proof_format}' not found in record, falling back to 'anoncreds'"
70	)
NEW UNCOV 71	proof_format = "anoncreds"	×
72	# If neither found, we stick with settings.ACAPY_PROOF_FORMAT and let it likely fail below
73
74	logger.info(	1✔
75	"Extracted requested attributes from presentation request",
76	requested_attributes=auth_session.presentation_exchange["pres_request"][
77	proof_format
78	]["requested_attributes"],
79	)
80
81	referent: str
82	requested_attr: ReqAttr
83	try:	1✔
84	for referent, requested_attrdict in auth_session.presentation_exchange[	1✔
85	"pres_request"
86	][proof_format]["requested_attributes"].items():
87	requested_attr = ReqAttr(**requested_attrdict)	1✔
88	logger.debug(	1✔
89	f"Processing referent: {referent}, requested_attr: {requested_attr}"
90	)
91	revealed_attrs: dict[str, RevealedAttribute] = (	1✔
92	auth_session.presentation_exchange["pres"][proof_format][
93	"requested_proof"
94	]["revealed_attr_groups"]
95	)
96	logger.debug(f"revealed_attrs: {revealed_attrs}")	1✔
97	# loop through each value and put it in token as a claim
98	for attr_name in requested_attr.names:	1✔
99	logger.debug(f"AttrName: {attr_name}")	1✔
100	presentation_claims[attr_name] = Claim(	1✔
101	type=attr_name,
102	value=revealed_attrs[referent]["values"][attr_name]["raw"],
103	)
104	logger.debug(f"Compiled presentation_claims: {presentation_claims}")	1✔
105	except Exception as err:	1✔
106	logger.error(	1✔
107	f"An exception occurred while extracting the proof claims: {err}"
108	)
109	raise RuntimeError(err)	1✔
110
111	proof_claims = json.dumps(	1✔
112	{c.type: c.value for c in presentation_claims.values()}
113	)
114	# look at all presentation_claims for one
115	# matching the configured subject_identifier, if any
116	sub_id_claim = presentation_claims.get(ver_config.subject_identifier)	1✔
117
118	pres_req_conf_id_suffix = (	1✔
119	f"@{auth_session.request_parameters['pres_req_conf_id']}"
120	)
121
122	if sub_id_claim:	1✔
123	# add sub and append presentation_claims
124	oidc_claims.append(	1✔
125	Claim(
126	type="sub",
127	value=sub_id_claim.value + pres_req_conf_id_suffix,
128	)
129	)
130
131	elif ver_config.generate_consistent_identifier:	1✔
132	# Do not create a sub based on the proof claims if the
133	# user requests a generated identifier
134	# Generate a SHA256 hash of the canonicaljson encoded proof_claims
135	encoded_json: bytes = canonicaljson.encode_canonical_json(proof_claims)	1✔
136	sha256_hash = hashlib.sha256(	1✔
137	encoded_json + pres_req_conf_id_suffix.encode()
138	).hexdigest()
139	oidc_claims.append(	1✔
140	Claim(
141	type="sub",
142	value=sha256_hash,
143	)
144	)
145
146	result = {c.type: c.value for c in oidc_claims}	1✔
147	result[PROOF_CLAIMS_ATTRIBUTE_NAME] = proof_claims	1✔
148
149	# TODO: Remove after full transistion to v2.0
150	# Add the presentation claims to the result as keys
151	# for backwards compatibility [v1.0]
152	if ver_config.include_v1_attributes:	1✔
153	for key, value in presentation_claims.items():	1✔
154	result[key] = value.value	1✔
155
156	return result	1✔
157
158	# TODO: Determine if this is useful to keep, and remove it if it's not.
159	# It is currently unused.
160	# renames and calculates dict members appropriate to
161	# https://openid.net/specs/openid-connect-core-1_0.html#IDToken
162	# and
163	# https://github.com/OpenIDC/pyoidc/blob/26ea5121239dad03c5c5551cca149cb984df1ec9/src/oic/oic/message.py#L720
164	def idtoken_dict(self, nonce: str) -> dict:	1✔
165	"""Converts oidc claims to IdToken attribute names"""
166
167	result = {} # nest VC attribute claims under the key=pres_req_conf_id	1✔
168
169	# for type, value in self.claims.items():
170	# result[type] = value
171
172	result["exp"] = int(round(datetime.now().timestamp())) + self.lifetime	1✔
173	result["aud"] = self.audiences	1✔
174	result["nonce"] = nonce	1✔
175
176	result.update(self.claims)	1✔
177
178	# identify if any standardclaims were provided in the proof and return
179	# them at the top level.
180	# https://openid.net/specs/openid-connect-core-1_0.html#StandardClaims
181
182	# make copy of dict
183	r2 = result.copy()	1✔
184	# add nested values to top level
185	r2.update(json.loads(self.claims[PROOF_CLAIMS_ATTRIBUTE_NAME]))	1✔
186	# only keep ones that match the OpenIDschema
187	r2 = {	1✔
188	key: r2[key]
189	for key in set(r2.keys()).intersection(set(OpenIDSchema().c_param.keys()))
190	}
191
192	# verify with library schema
193	standard_claims = OpenIDSchema().from_dict(r2)	1✔
194	standard_claims.verify()	1✔
195	# add to the top level of the dict.
196	for key, value in standard_claims.to_dict().items():	1✔
197	result[key] = value	1✔
198
199	return result	1✔

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

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