Skip to content

Commit 206095a

Browse files
PascalDRPascalDR
committed
Merge branch 'poc/openid4vci' of https://github.com/italia/eudi-wallet-it-python into poc/openid4vci
2 parents 3f954fe + 6349ba7 commit 206095a

5 files changed

Lines changed: 184 additions & 22 deletions

File tree

‎pyeudiw/openid4vci/endpoints/status_list_endpoint.py‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,8 @@ def endpoint(self, context: Context):
8080
self._log_error(
8181
self.__class__.__name__,
8282
f"unexpected accept header {accept_header} ")
83-
raise InvalidRequestException(f"Invalid accept header {accept_header}")
83+
raise InvalidRequestException(
84+
f"{'Invalid accept header' if accept_header is not None else 'Missing accept header'}")
8485
except (InvalidRequestException, InvalidScopeException) as e:
8586
return self._handle_400(context, e.message, e)
8687
except Exception as e:
@@ -99,18 +100,28 @@ def _build_status_list_payload(self):
99100
status_path = status_path.lstrip("/")
100101
iat = iat_now()
101102
credentials = self._db_credential_engine.get_all_sorted_by_incremental_id()
103+
if not credentials or len(credentials) == 0:
104+
lst = ''
105+
else:
106+
bit_bytes = array_to_bitstring(credentials)
107+
lst = bin(int.from_bytes(bit_bytes, 'big'))[2:].zfill(len(credentials))
102108
return {
103109
"exp": iat + self.status_list.exp,
104110
"iat": iat,
105111
"status_list": {
106112
"bits": 1,
107-
"lst": array_to_bitstring(credentials)
113+
"lst": lst
108114
},
109115
"sub": f"{self._backend_url}/{status_path}/1",
110116
"ttl": self.status_list.ttl
111117
}
112118

113119
def _validate_configs(self):
120+
cred_config = self.config_utils.get_credential_configurations()
121+
self._validate_required_configs([
122+
("credential_configurations", cred_config),
123+
])
124+
114125
status_list = self.config_utils.get_credential_configurations().status_list
115126
self._validate_required_configs([
116127
("credential_configurations.status_list", status_list),

‎pyeudiw/status_list/__init__.py‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,3 @@
1-
from binascii import unhexlify
21
import zlib
32
from binascii import hexlify
43
from binascii import unhexlify
@@ -198,10 +197,12 @@ def array_to_bitstring(status_array: list[dict], bit_size: int = 1) -> bytes:
198197
bitstring: int = 0
199198
for status in status_array:
200199
if status["revoked"]:
201-
bitstring |= 1 << (status["incremental_id"] - 1)
202-
elif not status["revoked"]:
203-
bitstring &= ~(1 << (status["incremental_id"] - 1))
200+
# Set bit to 1 if revoked
201+
bitstring |= 1 << (len(status_array) - status["incremental_id"])
202+
else:
203+
# Clear bit to 0 if not revoked
204+
bitstring &= ~(1 << (len(status_array) - status["incremental_id"]))
204205

205206
bit_length = len(status_array)
206207
byte_length = (bit_length + 7) // 8
207-
return bitstring.to_bytes(byte_length, byteorder='big', signed=False)
208+
return bitstring.to_bytes(byte_length, byteorder='big', signed=False)

‎pyeudiw/tests/openid4vci/endpoints/test_status_list_endpoint.py‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
from copy import deepcopy
2+
from unittest.mock import patch
3+
4+
import jwt
5+
import pytest
6+
from satosa.context import Context
7+
8+
from pyeudiw.openid4vci.endpoints.status_list_endpoint import StatusListHandler
9+
from pyeudiw.tests.openid4vci.endpoints.endpoints_test import (
10+
do_test_invalid_request_method,
11+
do_test_missing_configurations_raises,
12+
do_test_invalid_content_type, assert_invalid_request_application_json
13+
)
14+
from pyeudiw.tests.openid4vci.mock_openid4vci import (
15+
INVALID_CONTENT_TYPES_NOT_APPLICATION_JSON,
16+
INVALID_METHOD_FOR_GET_REQ,
17+
MOCK_STATUS_LIST_CONFIG,
18+
MOCK_PYEUDIW_FRONTEND_CONFIG,
19+
MOCK_INTERNAL_ATTRIBUTES,
20+
MOCK_NAME,
21+
MOCK_BASE_URL,
22+
get_mocked_satosa_context,
23+
mock_deserialized_overridable,
24+
REMOVE
25+
)
26+
from pyeudiw.tools.content_type import (
27+
APPLICATION_JSON,
28+
ACCEPT_HEADER,
29+
STATUS_LIST_JWT, STATUS_LIST_CWT
30+
)
31+
32+
_BASE_PATH = "pyeudiw.openid4vci.endpoints.status_list_endpoint"
33+
34+
@pytest.fixture
35+
def status_list_handler() -> StatusListHandler:
36+
with (patch(f"{_BASE_PATH}.UserCredentialEngine") as user_cred_eng_class,
37+
patch("pyeudiw.storage.user_credential_db_engine.CredentialStorage") as credential_storage_mock):
38+
user_cred_eng_class.db_user_storage_engine = credential_storage_mock.return_value
39+
handler = StatusListHandler(MOCK_PYEUDIW_FRONTEND_CONFIG, MOCK_INTERNAL_ATTRIBUTES, MOCK_BASE_URL, MOCK_NAME)
40+
return handler
41+
42+
@pytest.fixture()
43+
def context() -> Context:
44+
return get_mocked_satosa_context(method="GET", content_type=APPLICATION_JSON)
45+
46+
@pytest.mark.parametrize("method", INVALID_METHOD_FOR_GET_REQ)
47+
def test_invalid_request_method(status_list_handler, context, method):
48+
do_test_invalid_request_method(status_list_handler, context, method)
49+
50+
@pytest.mark.parametrize("content_type", INVALID_CONTENT_TYPES_NOT_APPLICATION_JSON)
51+
def test_invalid_content_type(status_list_handler, context, content_type):
52+
do_test_invalid_content_type(status_list_handler, context, content_type)
53+
54+
@pytest.mark.parametrize("config, missing_fields", [
55+
(mock_deserialized_overridable(MOCK_PYEUDIW_FRONTEND_CONFIG, {"credential_configurations": REMOVE}), ["credential_configurations"]),
56+
(mock_deserialized_overridable(MOCK_PYEUDIW_FRONTEND_CONFIG, {"credential_configurations.status_list": REMOVE}), ["credential_configurations.status_list"]),
57+
(mock_deserialized_overridable(MOCK_PYEUDIW_FRONTEND_CONFIG, {"credential_configurations.status_list.exp": REMOVE}), ["credential_configurations.status_list.exp"]),
58+
(mock_deserialized_overridable(MOCK_PYEUDIW_FRONTEND_CONFIG, {"credential_configurations.status_list.path": REMOVE}), ["credential_configurations.status_list.path"]),
59+
(mock_deserialized_overridable(MOCK_PYEUDIW_FRONTEND_CONFIG, {"credential_configurations.status_list.ttl": REMOVE}), ["credential_configurations.status_list.ttl"]),
60+
])
61+
def test_missing_configurations_raises(config, missing_fields):
62+
do_test_missing_configurations_raises(StatusListHandler, config, missing_fields)
63+
64+
@pytest.mark.parametrize("accept_header, error_desc", [
65+
(None, "Missing accept header"),
66+
("", "Missing accept header"),
67+
(" ", "Invalid accept header"),
68+
("accept-random-value", "Invalid accept header"),
69+
])
70+
def test_invalid_accept_header(status_list_handler, context, accept_header, error_desc):
71+
ctx = deepcopy(context)
72+
if accept_header:
73+
ctx.http_headers[ACCEPT_HEADER] = accept_header
74+
assert_invalid_request_application_json(
75+
status_list_handler.endpoint(ctx), error_desc)
76+
77+
status_array = [
78+
{"incremental_id": 1, "revoked": False},
79+
{"incremental_id": 2, "revoked": True},
80+
{"incremental_id": 3, "revoked": False},
81+
{"incremental_id": 4, "revoked": True},
82+
{"incremental_id": 5, "revoked": False},
83+
]
84+
85+
def test_should_return_status_list_jwt_credentials(status_list_handler, context):
86+
should_return_status_list(status_list_handler, context, STATUS_LIST_JWT, status_array,
87+
{'bits': 1, 'lst': '01010'})
88+
89+
def test_should_return_status_list_jwt_without_credentials(status_list_handler, context):
90+
should_return_status_list(status_list_handler, context, STATUS_LIST_JWT, [],
91+
{'bits': 1, 'lst': ''})
92+
93+
def should_return_status_list(status_list_handler, context: Context, accept_header: str, status_list: list[dict],
94+
expected_status_list: dict):
95+
status_list_handler._db_credential_engine.get_all_sorted_by_incremental_id.return_value = status_list
96+
ctx = deepcopy(context)
97+
ctx.http_headers[ACCEPT_HEADER] = accept_header
98+
result = status_list_handler.endpoint(ctx)
99+
assert result.status == '200 OK'
100+
if accept_header == STATUS_LIST_JWT:
101+
credential = jwt.decode(result.message, options={"verify_signature": False})
102+
elif accept_header == STATUS_LIST_CWT:
103+
credential = {}
104+
else:
105+
pytest.fail(f"Unexpected accept header value: {accept_header}")
106+
assert credential is not None
107+
assert credential["sub"] == f'{MOCK_BASE_URL}/{MOCK_NAME}{MOCK_STATUS_LIST_CONFIG["path"]}/1'
108+
assert credential["ttl"] == MOCK_STATUS_LIST_CONFIG["ttl"]
109+
assert credential["exp"] - credential["iat"] == MOCK_STATUS_LIST_CONFIG["exp"]
110+
assert credential["status_list"] == expected_status_list

‎pyeudiw/tests/openid4vci/mock_openid4vci.py‎

Lines changed: 26 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -123,13 +123,14 @@
123123
}
124124
}
125125

126+
MOCK_STATUS_LIST_CONFIG = {
127+
"path": "/status",
128+
"exp": 90,
129+
"ttl": 43200,
130+
}
126131
MOCK_CREDENTIAL_CONFIGURATIONS = {
127132
"lookup_source": "openid4vci",
128-
"status_list": {
129-
"path": "/status",
130-
"exp": 90,
131-
"ttl": 43200,
132-
},
133+
"status_list": MOCK_STATUS_LIST_CONFIG,
133134
"credential_specification": {
134135
"dc_sd_jwt_mDL": {
135136
"template": """
@@ -450,13 +451,26 @@ def get_pyeudiw_frontend_config_with_openid_credential_issuer(openid_credential_
450451

451452
return MOCK_PYEUDIW_FRONTEND_CONFIG
452453

453-
REMOVE = object() # special value for remove object
454+
REMOVE = object() # special value to remove keys
455+
454456
def mock_deserialized_overridable(base: dict, overrides=None):
455-
result = base.copy()
457+
import copy
458+
def set_or_remove(d: dict, path: list[str], value):
459+
*parents, last = path
460+
current = d
461+
for p in parents:
462+
current = current.setdefault(p, {}) # crea i dict intermedi se mancanti
463+
464+
if value is REMOVE:
465+
current.pop(last, None)
466+
else:
467+
current[last] = value
468+
469+
result = copy.deepcopy(base)
470+
456471
if overrides:
457472
for k, v in overrides.items():
458-
if v is REMOVE:
459-
result.pop(k, None) # remove key if exist
460-
else:
461-
result[k] = v
462-
return result
473+
path = k.split(".")
474+
set_or_remove(result, path, v)
475+
476+
return result

‎pyeudiw/tools/content_type.py‎

Lines changed: 29 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -44,8 +44,7 @@ def get_content_type_header(headers: list[tuple[str, str]]) -> str | None:
4444
Returns:
4545
str | None: The value of the Content-Type header if present, None otherwise.
4646
"""
47-
return next((v for k, v in headers if k.lower() == CONTENT_TYPE_HEADER), None)
48-
47+
return _get_header(headers, CONTENT_TYPE_HEADER)
4948

5049
def get_accept_header(headers: list[tuple[str, str]]) -> str | None:
5150
"""
@@ -57,4 +56,31 @@ def get_accept_header(headers: list[tuple[str, str]]) -> str | None:
5756
Returns:
5857
str | None: The value of the Accept header if present, None otherwise.
5958
"""
60-
return next((v for k, v in headers if k.lower() == ACCEPT_HEADER), None)
59+
return _get_header(headers, ACCEPT_HEADER)
60+
61+
def _get_header(headers, key):
62+
"""
63+
Retrieve the value of a header from a collection of headers.
64+
65+
This function supports both dictionaries and lists of (key, value) pairs.
66+
Keys are matched case-insensitively.
67+
68+
:param headers: The headers collection. Can be a dictionary or a list of 2-element tuples/lists.
69+
:type headers: dict or list[tuple[str, str]]
70+
71+
:param key: The header name to search for.
72+
:type key: str
73+
74+
:return: The value associated with the given header key, or None if not found.
75+
:rtype: str or None
76+
"""
77+
if isinstance(headers, dict):
78+
return headers.get(key) or headers.get(key.lower())
79+
elif isinstance(headers, list):
80+
return next(
81+
(v for h in headers if isinstance(h, (tuple, list)) and len(h) == 2
82+
for k, v in [h] if k.lower() == key.lower()),
83+
None
84+
)
85+
return None
86+

0 commit comments

Comments
 (0)