[S] Change in pysim[master]: smdp: update tls certs
by Hoernchen
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/40881?usp=email )
Change subject: smdp: update tls certs
......................................................................
smdp: update tls certs
Old expired today, new from https://www.gsma.com/solutions-and-impact/technologies/esim/wp-content/uplo…
Change-Id: I585efe3360a0aac2a49a79d5fef2789dea2b169d
---
M smdpp-data/certs/DPtls/CERT_S_SM_DP2_TLS.der
M smdpp-data/certs/DPtls/CERT_S_SM_DP4_TLS.der
M smdpp-data/certs/DPtls/CERT_S_SM_DP8_TLS.der
M smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.der
M smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.pem
M smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.der
M smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.pem
7 files changed, 10 insertions(+), 10 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/81/40881/1
diff --git a/smdpp-data/certs/DPtls/CERT_S_SM_DP2_TLS.der b/smdpp-data/certs/DPtls/CERT_S_SM_DP2_TLS.der
index 79ef55d..4df1f5a 100644
--- a/smdpp-data/certs/DPtls/CERT_S_SM_DP2_TLS.der
+++ b/smdpp-data/certs/DPtls/CERT_S_SM_DP2_TLS.der
Binary files differ
diff --git a/smdpp-data/certs/DPtls/CERT_S_SM_DP4_TLS.der b/smdpp-data/certs/DPtls/CERT_S_SM_DP4_TLS.der
index eb1d606..3af9b61 100644
--- a/smdpp-data/certs/DPtls/CERT_S_SM_DP4_TLS.der
+++ b/smdpp-data/certs/DPtls/CERT_S_SM_DP4_TLS.der
Binary files differ
diff --git a/smdpp-data/certs/DPtls/CERT_S_SM_DP8_TLS.der b/smdpp-data/certs/DPtls/CERT_S_SM_DP8_TLS.der
index dec58b9..de8f5f3 100644
--- a/smdpp-data/certs/DPtls/CERT_S_SM_DP8_TLS.der
+++ b/smdpp-data/certs/DPtls/CERT_S_SM_DP8_TLS.der
Binary files differ
diff --git a/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.der b/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.der
index 48f8422..9556049 100644
--- a/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.der
+++ b/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.der
Binary files differ
diff --git a/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.pem b/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.pem
index 53e7d0b..b4258fa 100644
--- a/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.pem
+++ b/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.pem
@@ -1,7 +1,7 @@
-----BEGIN CERTIFICATE-----
-MIICgjCCAimgAwIBAgIBCTAKBggqhkjOPQQDAjBEMRAwDgYDVQQDDAdUZXN0IENJ
+MIICgzCCAimgAwIBAgIBCTAKBggqhkjOPQQDAjBEMRAwDgYDVQQDDAdUZXN0IENJ
MREwDwYDVQQLDAhURVNUQ0VSVDEQMA4GA1UECgwHUlNQVEVTVDELMAkGA1UEBhMC
-SVQwHhcNMjQwNzA5MTUyOTM2WhcNMjUwODExMTUyOTM2WjAzMQ0wCwYDVQQKDARB
+SVQwHhcNMjUwNjMwMTMxNDM4WhcNMjYwODAyMTMxNDM4WjAzMQ0wCwYDVQQKDARB
Q01FMSIwIAYDVQQDDBl0ZXN0c21kcHBsdXMxLmV4YW1wbGUuY29tMFowFAYHKoZI
zj0CAQYJKyQDAwIIAQEHA0IABEwizNgsjQIh+dhUO3LhB7zJ/ZBU1mx1wOt0p73n
MOdhjvZbJwteguQ6eW+N7guvivvrilNiU3oC/WXHnkEZa7WjggEaMIIBFjAOBgNV
@@ -10,7 +10,7 @@
i50wHwYDVR0jBBgwFoAUwLxwujaSnUO0Z/9XVwUw5Xq4/NgwKQYDVR0RBCIwIIIZ
dGVzdHNtZHBwbHVzMS5leGFtcGxlLmNvbYgDiDcKMGEGA1UdHwRaMFgwKqAooCaG
JGh0dHA6Ly9jaS50ZXN0LmV4YW1wbGUuY29tL0NSTC1BLmNybDAqoCigJoYkaHR0
-cDovL2NpLnRlc3QuZXhhbXBsZS5jb20vQ1JMLUIuY3JsMAoGCCqGSM49BAMCA0cA
-MEQCIHHmXEy9mgudh/VbK0hJwmX7eOgbvHLnlujrpQzvUd4uAiBFVJgSdzYvrmJ9
-5yeIvmjHwxSMBgQp2dde7OtdVEK8Kw==
+cDovL2NpLnRlc3QuZXhhbXBsZS5jb20vQ1JMLUIuY3JsMAoGCCqGSM49BAMCA0gA
+MEUCIQCfaGcMk+kuSJsbIyRPWttwWNftwQdHCQuu346PaiA2FAIgUrqhPw2um9gV
+C+eWHaXio7WQh5L6VgLZzNifTQcldD4=
-----END CERTIFICATE-----
diff --git a/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.der b/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.der
index b819e77..ab03f3d 100644
--- a/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.der
+++ b/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.der
Binary files differ
diff --git a/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.pem b/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.pem
index c94d612..3d6dad7 100644
--- a/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.pem
+++ b/smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.pem
@@ -1,7 +1,7 @@
-----BEGIN CERTIFICATE-----
-MIICgzCCAiigAwIBAgIBCTAKBggqhkjOPQQDAjBEMRAwDgYDVQQDDAdUZXN0IENJ
+MIICgjCCAiigAwIBAgIBCTAKBggqhkjOPQQDAjBEMRAwDgYDVQQDDAdUZXN0IENJ
MREwDwYDVQQLDAhURVNUQ0VSVDEQMA4GA1UECgwHUlNQVEVTVDELMAkGA1UEBhMC
-SVQwHhcNMjQwNzA5MTUyODMzWhcNMjUwODExMTUyODMzWjAzMQ0wCwYDVQQKDARB
+SVQwHhcNMjUwNjMwMTMxNDM4WhcNMjYwODAyMTMxNDM4WjAzMQ0wCwYDVQQKDARB
Q01FMSIwIAYDVQQDDBl0ZXN0c21kcHBsdXMxLmV4YW1wbGUuY29tMFkwEwYHKoZI
zj0CAQYIKoZIzj0DAQcDQgAEKCQwdc6O/R+uZ2g5QH2ybkzLQ3CUYhybOWEz8bJL
tQG4/k6yTT4NOS8lP28blGJws8opLjTbb3qHs6X2rJRfCKOCARowggEWMA4GA1Ud
@@ -10,7 +10,7 @@
qTAfBgNVHSMEGDAWgBT1QXK9+YqV1ly+uIo4ocEdgAqFwzApBgNVHREEIjAgghl0
ZXN0c21kcHBsdXMxLmV4YW1wbGUuY29tiAOINwowYQYDVR0fBFowWDAqoCigJoYk
aHR0cDovL2NpLnRlc3QuZXhhbXBsZS5jb20vQ1JMLUEuY3JsMCqgKKAmhiRodHRw
-Oi8vY2kudGVzdC5leGFtcGxlLmNvbS9DUkwtQi5jcmwwCgYIKoZIzj0EAwIDSQAw
-RgIhAL1qQ/cnrCZC7UnnLJ8WeK+0aWUJFWh1cOlBEzw0NlTVAiEA25Vf4WHzwmJi
-zkARzxJ1qB0qfBofuJrtfPM4gNJ4Quw=
+Oi8vY2kudGVzdC5leGFtcGxlLmNvbS9DUkwtQi5jcmwwCgYIKoZIzj0EAwIDSAAw
+RQIhAL+1lp/hGsj87/5RqOX2u3hS/VSftDN7EPrHJJFnTXLRAiBVxemKIKmC7+W1
++RsTY5I51R+Cyoq4l5TEU49eplo5bw==
-----END CERTIFICATE-----
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/40881?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I585efe3360a0aac2a49a79d5fef2789dea2b169d
Gerrit-Change-Number: 40881
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
3 weeks, 5 days
- 1
- 0
[XS] Change in pysim[master]: bsp: fix maxpayloadsize
by Hoernchen
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/40882?usp=email )
Change subject: bsp: fix maxpayloadsize
......................................................................
bsp: fix maxpayloadsize
Change-Id: I08f544877b79681ad1f758a1ca31c292eae9f868
---
M pySim/esim/bsp.py
1 file changed, 1 insertion(+), 1 deletion(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/82/40882/1
diff --git a/pySim/esim/bsp.py b/pySim/esim/bsp.py
index 70e4a1f..cd4114f 100644
--- a/pySim/esim/bsp.py
+++ b/pySim/esim/bsp.py
@@ -282,7 +282,7 @@
def mac_only_one(self, tag: int, plaintext: bytes) -> bytes:
"""MAC a single plaintext TLV. Returns the protected ciphertext."""
assert tag <= 255
- assert len(plaintext) < self.max_payload_size
+ assert len(plaintext) <= self.max_payload_size
maced = self.m_algo.auth(tag, plaintext)
# The data block counter for ICV calculation is incremented also for each segment with C-MAC only.
self.c_algo.block_nr += 1
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/40882?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I08f544877b79681ad1f758a1ca31c292eae9f868
Gerrit-Change-Number: 40882
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
3 weeks, 5 days
- 1
- 0
[L] Change in pysim[master]: smdpp: cc support, proper checks, proper check order, test mode
by Hoernchen
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/40880?usp=email )
Change subject: smdpp: cc support, proper checks, proper check order, test mode
......................................................................
smdpp: cc support, proper checks, proper check order, test mode
Change-Id: Ie4db65594b2eaf6c95ffc5e73ff9ae61c4d9d3a3
XX cc support
Change-Id: I7b22fe4eecb41d0f98e28a1d9390cd7fc8c5784c
---
M osmo-smdpp.py
M pySim/esim/x509_cert.py
A pySim/esim/x509_err.py
3 files changed, 803 insertions(+), 78 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/80/40880/1
diff --git a/osmo-smdpp.py b/osmo-smdpp.py
index 94e64e8..69fae15 100755
--- a/osmo-smdpp.py
+++ b/osmo-smdpp.py
@@ -110,6 +110,7 @@
from cryptography.hazmat.primitives import hashes # noqa: E402
from cryptography.hazmat.primitives.asymmetric import ec, dh # noqa: E402
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat, PrivateFormat, NoEncryption, ParameterFormat # noqa: E402
+from cryptography.x509.oid import ExtensionOID, NameOID # noqa: E402
from pathlib import Path # noqa: E402
import json # noqa: E402
import sys # noqa: E402
@@ -121,6 +122,7 @@
from pprint import pprint as pp # noqa: E402
import base64 # noqa: E402
+import time # noqa: E402
from base64 import b64decode # noqa: E402
from klein import Klein # noqa: E402
from twisted.web.iweb import IRequest # noqa: E402
@@ -131,7 +133,11 @@
from pySim.esim import saip, PMO # noqa: E402
from pySim.esim.es8p import ProfileMetadata,UnprotectedProfilePackage,ProtectedProfilePackage,BoundProfilePackage,BspInstance # noqa: E402
from pySim.esim.x509_cert import oid, cert_policy_has_oid, cert_get_auth_key_id # noqa: E402
-from pySim.esim.x509_cert import CertAndPrivkey, CertificateSet, cert_get_subject_key_id, VerifyError # noqa: E402
+from cryptography.x509 import ExtensionNotFound # noqa: E402
+from pySim.esim.x509_cert import CertAndPrivkey, CertificateSet, cert_get_subject_key_id # noqa: E402
+from pySim.esim import x509_err # noqa: E402
+from datetime import datetime, timezone # noqa: E402
+import hashlib # noqa: E402
import logging # noqa: E402
logger = logging.getLogger(__name__)
@@ -344,6 +350,141 @@
return encode_dss_signature(r, s)
+def compute_confirmation_code_hash(confirmation_code: str, transaction_id: bytes) -> bytes:
+ """Compute confirmation code hash according to SGP.22 specification.
+ Hashed Confirmation Code = SHA256(SHA256(Confirmation Code) | TransactionID)
+ """
+
+ # Convert confirmation code from hex string to bytes (like EID handling)
+ cc_bytes = h2b(confirmation_code)
+
+ # Step 1: SHA256(Confirmation Code)
+ first_hash = hashlib.sha256(cc_bytes).digest()
+
+ # Step 2: SHA256(SHA256(CC) | TransactionID)
+ return hashlib.sha256(first_hash + transaction_id).digest()
+
+
+def validate_eum_certificate(eum_cert: x509.Certificate) -> None:
+ """Validate EUM certificate according to SGP.22 requirements.
+ Raises ApiError with appropriate error codes for different validation failures."""
+
+ # Check KeyUsage extension
+ try:
+ key_usage = eum_cert.extensions.get_extension_for_oid(ExtensionOID.KEY_USAGE)
+ # EUM certificates are intermediate CAs that sign eUICC certificates, so they need keyCertSign
+ if not key_usage.value.key_cert_sign:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+ # Check for critical flag
+ if not key_usage.critical:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+ except x509.ExtensionNotFound:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+
+ # ExtendedKeyUsage not present in EUM certificates as they are intermediate CAs (?)
+
+ # Check Certificate Policies
+ try:
+ cert_policies = eum_cert.extensions.get_extension_for_oid(ExtensionOID.CERTIFICATE_POLICIES)
+ # Check for required EUM policy OIDs
+ has_valid_policy = False
+ for policy in cert_policies.value:
+ policy_oid = policy.policy_identifier.dotted_string
+ # EUM policies: 2.23.146.1.2.1.2 (old) or 2.23.146.1.2.1.0.0.0 (new)
+ if policy_oid in ['2.23.146.1.2.1.2', '2.23.146.1.2.1.0.0.0']:
+ has_valid_policy = True
+ break
+ if not has_valid_policy:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+ except x509.ExtensionNotFound:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+
+ # Check BasicConstraints - EUM is an intermediate CA so it should have CA:TRUE (?)
+ try:
+ basic_constraints = eum_cert.extensions.get_extension_for_oid(ExtensionOID.BASIC_CONSTRAINTS)
+ if not basic_constraints.value.ca:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+ # pathLenConstraint of 0 means it can only sign end-entity certificates
+ if basic_constraints.value.path_length is not None and basic_constraints.value.path_length != 0:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+ except x509.ExtensionNotFound:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+
+
+def validate_euicc_certificate(euicc_cert: x509.Certificate) -> None:
+ """Validate eUICC certificate according to SGP.22 requirements.
+ Raises ApiError with appropriate error codes for different validation failures."""
+
+ # Check KeyUsage extension
+ try:
+ key_usage = euicc_cert.extensions.get_extension_for_oid(ExtensionOID.KEY_USAGE)
+ if not key_usage.value.digital_signature:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ if key_usage.value.key_cert_sign:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ except x509.ExtensionNotFound:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+
+ # ExtendedKeyUsage not present in eUICC certificates
+
+ # Check Certificate Policies
+ try:
+ cert_policies = euicc_cert.extensions.get_extension_for_oid(ExtensionOID.CERTIFICATE_POLICIES)
+ # Check for required eUICC policy OIDs
+ has_valid_policy = False
+ for policy in cert_policies.value:
+ policy_oid = policy.policy_identifier.dotted_string
+ # eUICC policies would be under GSMA arc
+ if policy_oid.startswith('2.23.146.1.2.1'):
+ has_valid_policy = True
+ break
+ if not has_valid_policy:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ except x509.ExtensionNotFound:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+
+ # Check Subject fields
+ subject = euicc_cert.subject
+
+ # Verify Organization field
+ try:
+ org_attrs = subject.get_attributes_for_oid(NameOID.ORGANIZATION_NAME)
+ if not org_attrs:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ except Exception:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+
+ # Verify SerialNumber field (contains EID)
+ try:
+ serial_attrs = subject.get_attributes_for_oid(NameOID.SERIAL_NUMBER)
+ if not serial_attrs:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ eid = serial_attrs[0].value
+ if len(eid) != 32 or not all(c in '0123456789ABCDEFabcdef' for c in eid):
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ except Exception as e:
+ if not isinstance(e, ApiError):
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+
+
class ApiError(Exception):
def __init__(self, subject_code: str, reason_code: str, message: Optional[str] = None,
subject_id: Optional[str] = None):
@@ -401,14 +542,19 @@
cs = CertificateSet(ci_cert)
cs.verify_cert_chain(self.dp_auth.cert)
return True
- except VerifyError:
+ except x509_err.VerifyError:
continue
return False
- def __init__(self, server_hostname: str, ci_certs_path: str, common_cert_path: str, use_brainpool: bool = False, in_memory: bool = False):
+ def __init__(self, server_hostname: str, ci_certs_path: str, common_cert_path: str, use_brainpool: bool = False, in_memory: bool = False, test_mode: bool = False):
self.server_hostname = server_hostname
self.upp_dir = os.path.realpath(os.path.join(DATA_DIR, 'upp'))
self.ci_certs = self.load_certs_from_path(ci_certs_path)
+ self.test_mode = test_mode
+ self.confirmation_codes = {
+ "CC_REQUIRED_TEST": "12345678" # Special matchingId for confirmation code tests
+ } if test_mode else {}
+
# load DPauth cert + key
self.dp_auth = CertAndPrivkey(oid.id_rspRole_dp_auth_v2)
cert_dir = common_cert_path
@@ -435,6 +581,230 @@
db_path = os.path.join(DATA_DIR, f"sm-dp-sessions-{session_db_suffix}")
self.rss = rsp.RspSessionStore(filename=db_path, in_memory=False)
logger.info(f"Using file-based session storage: {db_path}")
+ self.otpk_mapping = {} # Maps euicc_otpk -> (smdp_ot, smdp_otpk) for retry scenarios
+
+ # Initialize profile configurations for test cases
+ if test_mode:
+ self._init_test_profiles()
+ else:
+ # Initialize empty profile dictionaries when not in test mode
+ self.activation_code_profiles = {}
+ self.event_based_profiles = {}
+ self.default_profiles = {}
+
+ def _init_test_profiles(self):
+ """Initialize test profiles for different use cases."""
+ # Activation code profiles
+ print("INIT: Initializing test profiles...")
+ self.activation_code_profiles = {
+ 'TEST123': {
+ 'matchingId': 'TEST123',
+ 'confirmationCode': '12345678', # 8-digit numeric code
+ 'iccid': '8900000000000000001F',
+ 'profileName': 'Test Profile 1',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': None
+ },
+ 'AC_NOT_RELEASED': {
+ 'matchingId': 'AC_NOT_RELEASED',
+ 'confirmationCode': '87654321',
+ 'iccid': '8900000000000000002F',
+ 'profileName': 'Not Released Profile',
+ 'state': 'not_released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': None
+ },
+ 'AC_WITH_CC': {
+ 'matchingId': 'AC_WITH_CC',
+ 'confirmationCode': '11223344',
+ 'iccid': '8900000000000000003F',
+ 'profileName': 'Profile Requiring CC',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': None
+ },
+ 'AC_NO_CC': {
+ 'matchingId': 'AC_NO_CC',
+ 'confirmationCode': None, # No confirmation code required
+ 'iccid': '8900000000000000004F',
+ 'profileName': 'Profile Without CC',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': None
+ },
+ # Error test profiles
+ 'AC_NO_ELIGIBLE': {
+ 'matchingId': 'AC_NO_ELIGIBLE',
+ 'confirmationCode': None,
+ 'iccid': '8900000000000000009F',
+ 'profileName': 'Profile with No Eligible Device',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': None,
+ 'device_requirements': {
+ 'min_memory_mb': 999999, # Impossible requirement
+ 'required_features': ['IMPOSSIBLE_FEATURE']
+ }
+ },
+ 'AC_EXPIRED': {
+ 'matchingId': 'AC_EXPIRED',
+ 'confirmationCode': None,
+ 'iccid': '8900000000000000010F',
+ 'profileName': 'Expired Download Order',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': '2020-01-01T00:00:00Z' # Expired
+ },
+ 'AC_MAX_RETRIES': {
+ 'matchingId': 'AC_MAX_RETRIES',
+ 'confirmationCode': None,
+ 'iccid': '8900000000000000011F',
+ 'profileName': 'Max Retries Exceeded',
+ 'state': 'released',
+ 'download_attempts': 5, # Already exceeded max attempts
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': None
+ },
+ 'AC_RESTRICTED_EID': {
+ 'matchingId': 'AC_RESTRICTED_EID',
+ 'confirmationCode': None,
+ 'iccid': '8900000000000000012F',
+ 'profileName': 'Profile Restricted to Different EID',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': '89999999999999999999999999999999', # Different EID
+ 'expiration': None
+ },
+ 'AC_OTHER_EID': {
+ 'matchingId': 'AC_OTHER_EID',
+ 'confirmationCode': None,
+ 'iccid': '8900000000000000013F',
+ 'profileName': 'Profile for Other EID',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': '89888888888888888888888888888888', # Different EID
+ 'expiration': None
+ },
+ 'MATCHING_ID_1': {
+ 'matchingId': 'MATCHING_ID_1',
+ 'confirmationCode': None,
+ 'iccid': '8900000000000000017F',
+ 'profileName': 'Test Activation Code Profile',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': '89049032123451234512345678901235',
+ 'expiration': None
+ },
+ 'CC_REQUIRED_TEST': {
+ 'matchingId': 'CC_REQUIRED_TEST',
+ 'confirmationCode': '12345678', # Requires confirmation code
+ 'iccid': '8900000000000000019F',
+ 'profileName': 'CC Required Test Profile',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': None
+ }
+ }
+
+ # SM-DS event-based profiles
+ self.event_based_profiles = {
+ 'EVENT_001': {
+ 'matchingId': 'EVENT_001',
+ 'confirmationCode': '55667788',
+ 'iccid': '8900000000000000005F',
+ 'profileName': 'Event-based Profile 1',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': None
+ },
+ 'UNMATCHED_EVENT': {
+ 'matchingId': 'UNMATCHED_EVENT',
+ 'confirmationCode': '99887766',
+ 'iccid': '8900000000000000006F',
+ 'profileName': 'Unmatched Event Profile',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': '89001012012341234012345678901224', # Different EID
+ 'expiration': None
+ },
+ 'EVENT_NORMAL': {
+ 'matchingId': 'EVENT_NORMAL',
+ 'confirmationCode': None,
+ 'iccid': '8900000000000000014F',
+ 'profileName': 'Normal SM-DS Event',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': None
+ },
+ 'EVENT_RESTRICTED': {
+ 'matchingId': 'EVENT_RESTRICTED',
+ 'confirmationCode': None,
+ 'iccid': '8900000000000000015F',
+ 'profileName': 'SM-DS Event Restricted to Different EID',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': '89777777777777777777777777777777', # Different EID
+ 'expiration': None
+ },
+ 'MATCHING_ID_EVENT': {
+ 'matchingId': 'MATCHING_ID_EVENT',
+ 'confirmationCode': None,
+ 'iccid': '8900000000000000018F',
+ 'profileName': 'Test Event-based Profile',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': '89049032123451234512345678901235',
+ 'expiration': None
+ }
+ }
+
+ # Default SM-DP+ profiles (associated with specific EIDs)
+ self.default_profiles = {
+ '89001012012341234012345678901234': { # Test EID
+ 'confirmationCode': '12345678',
+ 'iccid': '8900000000000000007F',
+ 'profileName': 'Default Profile for Test EID',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'expiration': None
+ },
+ '89049032123451234512345678901235': { # EID1 from test specs
+ 'confirmationCode': None,
+ 'iccid': '8900000000000000020F',
+ 'profileName': 'Default Profile for EID1',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'expiration': None
+ },
+ }
@app.handle_errors(ApiError)
def handle_apierror(self, request: IRequest, failure):
@@ -475,12 +845,12 @@
@app.route('/gsma/rsp2/es9plus/initiateAuthentication', methods=['POST'])
@rsp_api_wrapper
- def initiateAutentication(self, request: IRequest, content: dict) -> dict:
+ def initiateAuthentication(self, request: IRequest, content: dict) -> dict:
"""See ES9+ InitiateAuthentication SGP.22 Section 5.6.1"""
- # Verify that the received address matches its own SM-DP+ address, where the comparison SHALL be
- # case-insensitive. Otherwise, the SM-DP+ SHALL return a status code "SM-DP+ Address - Refused".
- if content['smdpAddress'] != self.server_hostname:
- raise ApiError('8.8.1', '3.8', 'Invalid SM-DP+ Address')
+ # SGP.22 v2.5 Section 5.6.1: Verify that the received address matches its own SM-DP+ address,
+ # where the comparison SHALL be case-insensitive.
+ if content['smdpAddress'].lower() != self.server_hostname.lower():
+ raise ApiError('8.8.1', '3.8', 'Invalid SM-DP+ Address') # SGP.22 Table 46
euiccChallenge = b64decode(content['euiccChallenge'])
if len(euiccChallenge) != 16:
@@ -489,7 +859,22 @@
euiccInfo1_bin = b64decode(content['euiccInfo1'])
euiccInfo1 = rsp.asn1.decode('EUICCInfo1', euiccInfo1_bin)
logger.debug("Rx euiccInfo1: %s" % euiccInfo1)
- #euiccInfo1['svn']
+
+ # Validate specification version (SVN)
+ svn = euiccInfo1.get('svn', b'\x02\x02\x00') # Default to v2.2.0 if not present
+ # Convert SVN bytes to version tuple (major, minor, revision)
+ svn_version = (svn[0], svn[1], svn[2])
+
+ # SM-DP+ supports versions from 2.0.0 to 2.3.x
+ min_version = (2, 0, 0)
+ max_version = (2, 3, 255) # Allow any 2.3.x version
+
+ if svn_version < min_version:
+ # SGP.22 Table 46: Specification Version Number - Unsupported
+ raise ApiError('8.8.3', '3.1', 'The Specification Version Number indicated by the eUICC is not supported by the SM-DP+')
+ elif svn_version[:2] > max_version[:2]: # Compare major.minor only for upper bound
+ # SGP.22 Table 46: Specification Version Number - Unsupported
+ raise ApiError('8.8.3', '3.1', 'The Specification Version Number indicated by the eUICC is not supported by the SM-DP+')
# TODO: If euiccCiPKIdListForSigningV3 is present ...
@@ -497,25 +882,36 @@
if 'euiccCiPKIdListForSigningV3' in euiccInfo1:
pkid_list = pkid_list + euiccInfo1['euiccCiPKIdListForSigningV3']
- # Validate that SM-DP+ supports certificate chains for verification
- verification_pkid_list = euiccInfo1.get('euiccCiPKIdListForVerification', [])
- if verification_pkid_list and not self.validate_certificate_chain_for_verification(verification_pkid_list):
- raise ApiError('8.8.4', '3.7', 'The SM-DP+ has no CERT.DPauth.SIG which chains to one of the eSIM CA Root CA Certificate with a Public Key supported by the eUICC')
-
- # verify it supports one of the keys indicated by euiccCiPKIdListForSigning
+ # First verify PKIDs for signing are supported (check this before verification PKIDs)
ci_cert = None
for x in pkid_list:
ci_cert = self.ci_get_cert_for_pkid(x)
+ if not ci_cert:
+ # Skip if no certificate found for this PKID
+ continue
# we already support multiple CI certificates but only one set of DPauth + DPpb keys. So we must
# make sure we choose a CI key-id which has issued both the eUICC as well as our own SM-DP side
# certs.
- if ci_cert and cert_get_subject_key_id(ci_cert) == self.dp_auth.get_authority_key_identifier().key_identifier:
+ try:
+ ci_subject_key_id = cert_get_subject_key_id(ci_cert)
+ except Exception:
+ # For CI certs, we need to get the subject key identifier differently
+ subject_key_ext = ci_cert.extensions.get_extension_for_oid(x509.oid.ExtensionOID.SUBJECT_KEY_IDENTIFIER)
+ ci_subject_key_id = subject_key_ext.value.key_identifier
+
+ if ci_subject_key_id == self.dp_auth.get_authority_key_identifier().key_identifier:
break
else:
ci_cert = None
if not ci_cert:
raise ApiError('8.8.2', '3.1', 'None of the proposed Public Key Identifiers is supported by the SM-DP+')
+ # After verifying signing PKIDs, validate certificate chains for verification
+ verification_pkid_list = euiccInfo1.get('euiccCiPKIdListForVerification', [])
+ if verification_pkid_list and not self.validate_certificate_chain_for_verification(verification_pkid_list):
+ # SGP.22 Table 46: SM-DP+ Certificate - Unavailable
+ raise ApiError('8.8.4', '3.7', 'The SM-DP+ has no CERT.DPAuth.ECDSA signed by one of the GSMA CI Public Key supported by the eUICC')
+
# Generate a TransactionID which is used to identify the ongoing RSP session. The TransactionID
# SHALL be unique within the scope and lifetime of each SM-DP+.
transactionId = uuid.uuid4().hex.upper()
@@ -584,69 +980,265 @@
euicc_cert = x509.load_der_x509_certificate(euiccCertificate_bin)
eum_cert = x509.load_der_x509_certificate(eumCertificate_bin)
- # Verify that the transactionId is known and relates to an ongoing RSP session. Otherwise, the SM-DP+
- # SHALL return a status code "TransactionId - Unknown"
+ # Verify that the transactionId is known and relates to an ongoing RSP session.
ss = self.rss.get(transactionId, None)
if ss is None:
- raise ApiError('8.10.1', '3.9', 'Unknown')
+ # SGP.22 Table 49: TransactionId - Unknown
+ raise ApiError('8.10.1', '3.9', 'The RSP session identified by the TransactionID is unknown')
ss.euicc_cert = euicc_cert
ss.eum_cert = eum_cert # TODO: do we need this in the state?
- # Verify that the Root Certificate of the eUICC certificate chain corresponds to the
- # euiccCiPKIdToBeUsed or TODO: euiccCiPKIdToBeUsedV3
- if cert_get_auth_key_id(eum_cert) != ss.ci_cert_id:
- raise ApiError('8.11.1', '3.9', 'Unknown')
+ # First verify that the Root Certificate of the eUICC certificate chain corresponds to the
+ # euiccCiPKIdToBeUsed - this check must come before detailed certificate validation
+ try:
+ eum_auth_key_id = cert_get_auth_key_id(eum_cert)
+ except (ExtensionNotFound, Exception) as e:
+ logger.error(f"Failed to get AuthorityKeyIdentifier from EUM certificate: {e}")
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+
+ if eum_auth_key_id != ss.ci_cert_id:
+ # SGP.22 Table 49: CI Public Key - Unknown
+ raise ApiError('8.11.1', '3.9', 'Unknown CI Public Key. The CI used by the EUM Certificate is not a trusted root for the SM-DP+')
+
+ # Certificate validation checks - only do after CI key verification
+ try:
+ # Check if certificates are valid (not expired, proper format, etc.)
+ now = datetime.now(timezone.utc)
+
+ # Check EUM certificate validity period
+ if eum_cert.not_valid_after_utc < now:
+ # SGP.22 Table 49: EUM Certificate - Expired
+ raise ApiError('8.1.2', '6.3', 'Certificate has expired')
+ if eum_cert.not_valid_before_utc > now:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+
+ # Check eUICC certificate validity period
+ if euicc_cert.not_valid_after_utc < now:
+ # SGP.22 Table 49: eUICC Certificate - Expired
+ raise ApiError('8.1.3', '6.3', 'Certificate has expired')
+ if euicc_cert.not_valid_before_utc > now:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+
+ # Perform detailed certificate validation
+ validate_eum_certificate(eum_cert)
+ validate_euicc_certificate(euicc_cert)
+
+ except ApiError:
+ raise # Re-raise our API errors
+ except Exception as e:
+ logger.error(f"Certificate validation error: {e}")
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
# Verify the validity of the eUICC certificate chain
- cs = CertificateSet(self.ci_get_cert_for_pkid(ss.ci_cert_id))
- cs.add_intermediate_cert(eum_cert)
- # TODO v3: otherCertsInChain
try:
+ cs = CertificateSet(self.ci_get_cert_for_pkid(ss.ci_cert_id))
+ cs.add_intermediate_cert(eum_cert)
+ # TODO v3: otherCertsInChain
cs.verify_cert_chain(euicc_cert)
- except VerifyError:
- raise ApiError('8.1.3', '6.1', 'Verification failed (certificate chain)')
- # raise ApiError('8.1.3', '6.3', 'Expired')
+ except x509_err.MissingIntermediateCert as e:
+ # Check if the missing certificate is the EUM cert
+ if hasattr(e, 'auth_key_id') and e.auth_key_id == b2h(cert_get_auth_key_id(eum_cert)):
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+ else:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ except x509_err.CertificateRevoked as e:
+ eum_serial = format(eum_cert.serial_number, 'X')
+ euicc_serial = format(euicc_cert.serial_number, 'X')
+
+ if e.cert_serial.upper() == eum_serial.upper():
+ # SGP.22 Table 49: EUM Certificate - Expired (revoked uses expired code)
+ raise ApiError('8.1.2', '6.3', 'Certificate has expired')
+ elif e.cert_serial.upper() == euicc_serial.upper():
+ # SGP.22 Table 49: eUICC Certificate - Expired (revoked uses expired code)
+ raise ApiError('8.1.3', '6.3', 'Certificate has expired')
+ except x509_err.MaxDepthExceeded:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ except x509_err.SignatureVerification as e:
+ # Check which certificate's signature failed
+ error_str = str(e)
+ if 'EUM' in error_str or 'intermediate' in error_str:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+ else:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ except x509_err.VerifyError as e:
+ # Generic certificate chain error for any other x509_err subclasses
+ error_str = str(e)
+ if 'EUM' in error_str or 'intermediate' in error_str:
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+ else:
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ except ValueError as e:
+ # This could be raised by add_intermediate_cert for invalid EUM cert
+ if 'intermediate certificate' in str(e):
+ # SGP.22 Table 49: EUM Certificate - Verification Failed
+ raise ApiError('8.1.2', '6.1', 'Certificate is invalid')
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
+ except Exception as e:
+ logger.error(f"Certificate chain verification error: {e}")
+ # Check if it's a missing CI certificate issue
+ if 'ci_cert_id' in str(e) or 'Unknown' in str(e):
+ # SGP.22 Table 49: CI Public Key - Unknown
+ raise ApiError('8.11.1', '3.9', 'Unknown CI Public Key. The CI used by the EUM Certificate is not a trusted root for the SM-DP+')
+ # SGP.22 Table 49: eUICC Certificate - Verification Failed
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
# Verify euiccSignature1 over euiccSigned1 using pubkey from euiccCertificate.
- # Otherwise, the SM-DP+ SHALL return a status code "eUICC - Verification failed"
if not self._ecdsa_verify(euicc_cert, euiccSignature1_bin, euiccSigned1_bin):
- raise ApiError('8.1', '6.1', 'Verification failed (euiccSignature1 over euiccSigned1)')
+ # SGP.22 Table 49: eUICC - Verification Failed
+ raise ApiError('8.1', '6.1', 'eUICC signature is invalid or serverChallenge is invalid')
ss.eid = ss.euicc_cert.subject.get_attributes_for_oid(x509.oid.NameOID.SERIAL_NUMBER)[0].value
logger.debug("EID (from eUICC cert): %s" % ss.eid)
# Verify EID is within permitted range of EUM certificate
if not validate_eid_range(ss.eid, eum_cert):
- raise ApiError('8.1.4', '6.1', 'EID is not within the permitted range of the EUM certificate')
+ # Use eUICC Certificate error since EID is from eUICC cert
+ raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
# Verify that the serverChallenge attached to the ongoing RSP session matches the
- # serverChallenge returned by the eUICC. Otherwise, the SM-DP+ SHALL return a status code "eUICC -
- # Verification failed".
+ # serverChallenge returned by the eUICC.
if euiccSigned1['serverChallenge'] != ss.serverChallenge:
- raise ApiError('8.1', '6.1', 'Verification failed (serverChallenge)')
+ # SGP.22 Table 49: eUICC - Verification Failed
+ raise ApiError('8.1', '6.1', 'eUICC signature is invalid or serverChallenge is invalid')
+
+ # Verify that the transactionId in euiccSigned1 matches the outer transactionId
+ if euiccSigned1['transactionId'] != h2b(transactionId):
+ # SGP.22 Table 49: TransactionId - Unknown
+ raise ApiError('8.10.1', '3.9', 'The RSP session identified by the TransactionID is unknown')
# If ctxParams1 contains a ctxParamsForCommonAuthentication data object, the SM-DP+ Shall [...]
- # TODO: We really do a very simplistic job here, this needs to be properly implemented later,
- # considering all the various cases, profile state, etc.
if euiccSigned1['ctxParams1'][0] == 'ctxParamsForCommonAuthentication':
cpca = euiccSigned1['ctxParams1'][1]
matchingId = cpca.get('matchingId', None)
- if not matchingId:
- # TODO: check if any pending profile downloads for the EID
- raise ApiError('8.2.6', '3.8', 'Refused')
- if matchingId:
- # look up profile based on matchingID. We simply check if a given file exists for now..
- path = os.path.join(self.upp_dir, matchingId) + '.der'
- # prevent directory traversal attack
- if os.path.commonprefix((os.path.realpath(path),self.upp_dir)) != self.upp_dir:
- raise ApiError('8.2.6', '3.8', 'Refused')
- if not os.path.isfile(path) or not os.access(path, os.R_OK):
- raise ApiError('8.2.6', '3.8', 'Refused')
- ss.matchingId = matchingId
- with open(path, 'rb') as f:
- pes = saip.ProfileElementSequence.from_der(f.read())
- iccid_str = b2h(pes.get_pe_for_type('header').decoded['iccid'])
+ logger.debug(f"Extracted matchingId from request: {matchingId}")
+
+ # Determine use case and find profile
+ profile_info = None
+ iccid_str = None
+
+ if not matchingId or matchingId == '':
+ # Default SM-DP+ address use case - check if EID has pending profile
+ logger.debug(f"Default SM-DP+ use case for EID: {ss.eid}")
+ if self.test_mode and ss.eid in self.default_profiles:
+ profile_info = self.default_profiles[ss.eid]
+ iccid_str = profile_info['iccid']
+ ss.matchingId = None # No matchingId for default case
+ elif not self.test_mode:
+ # In production mode, try to load a default profile from file system
+ # This is where real implementation would check database/backend
+ # SGP.22 Table 49: EID - Refused
+ raise ApiError('8.1.1', '3.8', 'EID doesn\'t match the expected value')
+ else:
+ # Test mode but no profile for this EID
+ # SGP.22 Table 49: EID - Refused
+ raise ApiError('8.1.1', '3.8', 'EID doesn\'t match the expected value')
+
+ elif self.test_mode and matchingId.startswith('EVENT_'):
+ # SM-DS event-based use case (test mode only)
+ logger.debug(f"SM-DS event use case with matchingId: {matchingId}")
+ if matchingId in self.event_based_profiles:
+ profile_info = self.event_based_profiles[matchingId]
+ # Check if profile is associated with specific EID
+ if profile_info.get('associated_eid') and profile_info['associated_eid'] != ss.eid:
+ # SGP.22 Table 49: EID - Refused
+ raise ApiError('8.1.1', '3.8', 'EID doesn\'t match the expected value')
+ iccid_str = profile_info['iccid']
+ ss.matchingId = matchingId
+ else:
+ # SGP.22 Table 49: MatchingID - Refused
+ raise ApiError('8.2.6', '3.8', 'MatchingID (AC_Token or EventID) is refused')
+
+ else:
+ # Activation code use case
+ logger.debug(f"Activation code use case with matchingId: {matchingId}")
+ logger.debug(f"Available activation codes: {list(self.activation_code_profiles.keys())}")
+ if self.test_mode and matchingId in self.activation_code_profiles:
+ profile_info = self.activation_code_profiles[matchingId]
+ # Check if profile is associated with specific EID
+ if profile_info.get('associated_eid') and profile_info['associated_eid'] != ss.eid:
+ # SGP.22 Table 49: EID - Refused
+ raise ApiError('8.1.1', '3.8', 'EID doesn\'t match the expected value')
+ iccid_str = profile_info['iccid']
+ ss.matchingId = matchingId
+ else:
+ path = os.path.join(self.upp_dir, matchingId) + '.der'
+ # prevent directory traversal attack
+ if os.path.commonprefix((os.path.realpath(path),self.upp_dir)) != self.upp_dir:
+ # SGP.22 Table 49: MatchingID - Refused
+ raise ApiError('8.2.6', '3.8', 'MatchingID (AC_Token or EventID) is refused')
+ if os.path.isfile(path) and os.access(path, os.R_OK):
+ with open(path, 'rb') as f:
+ pes = saip.ProfileElementSequence.from_der(f.read())
+ iccid_str = b2h(pes.get_pe_for_type('header').decoded['iccid'])
+ ss.matchingId = matchingId
+ # temporary profile info for legacy files
+ profile_info = {
+ 'confirmationCode': self.confirmation_codes.get(matchingId),
+ 'state': 'released',
+ 'profileName': matchingId
+ }
+ else:
+ # SGP.22 Table 49: MatchingID - Refused
+ raise ApiError('8.2.6', '3.8', 'MatchingID (AC_Token or EventID) is refused')
+
+ # Validate profile state and other conditions
+ if profile_info:
+ # Check if profile is released
+ if profile_info.get('state') == 'not_released':
+ # SGP.22 Table 49: Profile - Not allowed
+ raise ApiError('8.2', '1.2', 'Profile has not yet been released')
+
+ # Check for expired download order
+ if profile_info.get('expiration'):
+ exp_date = datetime.fromisoformat(profile_info['expiration'].replace('Z', '+00:00'))
+ if datetime.now(timezone.utc) > exp_date:
+ # SGP.22 Table 49: Download order - Time to Live Expired
+ raise ApiError('8.8.5', '4.10', 'The Download order has expired')
+
+ # Check maximum download attempts
+ if profile_info.get('download_attempts', 0) >= 3:
+ # SGP.22 Table 49: Download order - Maximum number of retries exceeded
+ raise ApiError('8.8.5', '6.4', 'The maximum number of retries for the Profile download order has been exceeded')
+
+ # Check device eligibility (for Test #15 - test mode only)
+ if self.test_mode and profile_info.get('device_requirements'):
+ # fixme real implementation would check against actual device capabilities
+ if profile_info['device_requirements'].get('min_memory_mb', 0) > 1000:
+ # SGP.22 Table 49: Profile Type - Stopped on warning
+ raise ApiError('8.2.5', '4.3', 'No eligible Profile for this eUICC/Device')
+
+ # Set confirmation code requirement
+ # Special handling: When matchingId is omitted or empty, don't require confirmation code
+ # per SGP.23 test sequences #14-#18
+ if self.test_mode and (not matchingId or matchingId == ''):
+ ss.ccRequiredFlag = False
+ logger.info("ccRequiredFlag=False for omitted/empty matchingId (default SM-DP+ use case)")
+ elif profile_info.get('confirmationCode'):
+ ss.ccRequiredFlag = True
+ ss.expected_confirmation_code = profile_info['confirmationCode']
+ logger.info(f"Set ccRequiredFlag=True with code: {ss.expected_confirmation_code}")
+ else:
+ ss.ccRequiredFlag = False
+ logger.info("ccRequiredFlag=False, no confirmation code required")
+
+ # Use profile name from profile_info if available
+ profile_name = profile_info.get('profileName', matchingId or 'DefaultProfile')
+ else:
+ # Should not happen if all cases are handled above
+ raise ApiError('8.2.6', '3.8', 'Profile not found')
else:
# there's currently no other option in the ctxParams1 choice, so this cannot happen
raise ApiError('1.3.1', '2.2', 'ctxParams1 missing mandatory ctxParamsForCommonAuthentication')
@@ -654,19 +1246,25 @@
# FIXME: we actually want to perform the profile binding herr, and read the profile metadat from the profile
# Put together profileMetadata + _bin
- ss.profileMetadata = ProfileMetadata(iccid_bin=h2b(swap_nibbles(iccid_str)), spn="OsmocomSPN", profile_name=matchingId)
+ ss.profileMetadata = ProfileMetadata(iccid_bin=h2b(swap_nibbles(iccid_str)), spn="OsmocomSPN", profile_name=profile_name)
# enable notifications for all operations
for event in ['enable', 'disable', 'delete']:
ss.profileMetadata.add_notification(event, self.server_hostname)
profileMetadata_bin = ss.profileMetadata.gen_store_metadata_request()
# Put together smdpSigned2 + _bin
+ cc_flag = getattr(ss, 'ccRequiredFlag', False)
+ logger.info(f"Setting ccRequiredFlag in SmdpSigned2: {cc_flag}")
+ logger.info(f"Session state has ccRequiredFlag: {hasattr(ss, 'ccRequiredFlag')}")
+ if hasattr(ss, 'ccRequiredFlag'):
+ logger.info(f"Session ccRequiredFlag value: {ss.ccRequiredFlag}")
smdpSigned2 = {
'transactionId': h2b(ss.transactionId),
- 'ccRequiredFlag': False, # whether the Confirmation Code is required
+ 'ccRequiredFlag': cc_flag, # whether the Confirmation Code is required
#'bppEuiccOtpk': None, # whether otPK.EUICC.ECKA already used for binding the BPP, tag '5F49'
}
smdpSigned2_bin = rsp.asn1.encode('SmdpSigned2', smdpSigned2)
+ logger.info(f"Encoded smdpSigned2 with ccRequiredFlag={cc_flag}")
ss.smdpSignature2_do = b'\x5f\x37\x40' + self.dp_pb.ecdsa_sign(smdpSigned2_bin + b'\x5f\x37\x40' + euiccSignature1_bin)
@@ -689,6 +1287,7 @@
# Verify that the received transactionId is known and relates to an ongoing RSP session
ss = self.rss.get(transactionId, None)
if not ss:
+ # SGP.22 Table 51: TransactionId - Unknown
raise ApiError('8.10.1', '3.9', 'The RSP session identified by the TransactionID is unknown')
prepDownloadResp_bin = b64decode(content['prepareDownloadResponse'])
@@ -699,7 +1298,9 @@
r_err = prepDownloadResp[1]
#r_err['transactionId']
#r_err['downloadErrorCode']
- raise ValueError("downloadResponseError %s" % r_err)
+ # Download error from eUICC - use generic eUICC verification failed
+ # SGP.22 Table 51: eUICC - Verification Failed
+ raise ApiError('8.1', '6.1', 'eUICC signature is invalid')
r_ok = prepDownloadResp[1]
@@ -707,11 +1308,13 @@
euiccSigned2 = r_ok['euiccSigned2']
euiccSigned2_bin = rsp.extract_euiccSigned2(prepDownloadResp_bin)
if not self._ecdsa_verify(ss.euicc_cert, r_ok['euiccSignature2'], euiccSigned2_bin + ss.smdpSignature2_do):
+ # SGP.22 Table 51: eUICC - Verification Failed
raise ApiError('8.1', '6.1', 'eUICC signature is invalid')
# not in spec: Verify that signed TransactionID is outer transaction ID
if h2b(transactionId) != euiccSigned2['transactionId']:
- raise ApiError('8.10.1', '3.9', 'The signed transactionId != outer transactionId')
+ # SGP.22 Table 51: TransactionId - Unknown
+ raise ApiError('8.10.1', '3.9', 'The RSP session identified by the TransactionID is unknown')
# store otPK.EUICC.ECKA in session state
ss.euicc_otpk = euiccSigned2['euiccOtpk']
@@ -720,11 +1323,30 @@
# Generate a one-time ECKA key pair (ot{PK,SK}.DP.ECKA) using the curve indicated by the Key Parameter
# Reference value of CERT.DPpb.ECDDSA
logger.debug("curve = %s" % self.dp_pb.get_curve())
- ss.smdp_ot = ec.generate_private_key(self.dp_pb.get_curve())
- # extract the public key in (hopefully) the right format for the ES8+ interface
- ss.smdp_otpk = ss.smdp_ot.public_key().public_bytes(Encoding.X962, PublicFormat.UncompressedPoint)
- logger.debug("smdpOtpk: %s" % b2h(ss.smdp_otpk))
- logger.debug("smdpOtsk: %s" % b2h(ss.smdp_ot.private_bytes(Encoding.DER, PrivateFormat.PKCS8, NoEncryption())))
+
+ # Check if we've seen this euicc_otpk before (retry scenario)
+ euicc_otpk_hex = b2h(euiccSigned2['euiccOtpk'])
+ if euicc_otpk_hex in self.otpk_mapping:
+ # Retry scenario - reuse existing keys
+ existing_data = self.otpk_mapping[euicc_otpk_hex]
+ ss.smdp_ot = existing_data['smdp_ot']
+ ss.smdp_otpk = existing_data['smdp_otpk']
+ logger.debug("Retry scenario detected - reusing existing smdp_otpk")
+ logger.debug("smdpOtpk (reused): %s" % b2h(ss.smdp_otpk))
+ else:
+ # Generate new keys
+ ss.smdp_ot = ec.generate_private_key(self.dp_pb.get_curve())
+ # extract the public key in (hopefully) the right format for the ES8+ interface
+ ss.smdp_otpk = ss.smdp_ot.public_key().public_bytes(Encoding.X962, PublicFormat.UncompressedPoint)
+ logger.debug("smdpOtpk: %s" % b2h(ss.smdp_otpk))
+ logger.debug("smdpOtsk: %s" % b2h(ss.smdp_ot.private_bytes(Encoding.DER, PrivateFormat.PKCS8, NoEncryption())))
+
+ # Store the mapping for retry scenarios
+ self.otpk_mapping[euicc_otpk_hex] = {
+ 'smdp_ot': ss.smdp_ot,
+ 'smdp_otpk': ss.smdp_otpk,
+ 'timestamp': time.time()
+ }
ss.host_id = b'mahlzeit'
@@ -733,9 +1355,26 @@
ss.shared_secret = ss.smdp_ot.exchange(ec.ECDH(), euicc_public_key)
logger.debug("shared_secret: %s" % b2h(ss.shared_secret))
- # TODO: Check if this order requires a Confirmation Code verification
+ # Check if this order requires a Confirmation Code verification
+ if getattr(ss, 'ccRequiredFlag', False):
+ if 'hashCc' not in euiccSigned2:
+ # SGP.22 Table 51: Confirmation Code - Mandatory Element Missing
+ raise ApiError('8.2.7', '2.2', 'Confirmation Code is missing')
- # Perform actual protection + binding of profile package (or return pre-bound one)
+ received_hash = euiccSigned2['hashCc']
+ expected_hash = compute_confirmation_code_hash(
+ getattr(ss, 'expected_confirmation_code', ''),
+ h2b(transactionId)
+ )
+
+ if received_hash != expected_hash:
+ logger.debug("Confirmation code verification failed")
+ logger.debug(f"Expected hash: {b2h(expected_hash)}")
+ logger.debug(f"Received hash: {b2h(received_hash)}")
+ # SGP.22 Table 51: Confirmation Code - Refused
+ raise ApiError('8.2.7', '3.8', 'Confirmation Code is refused')
+
+ # Perform actual protection + binding of profile package (or return pre-bound one)
with open(os.path.join(self.upp_dir, ss.matchingId)+'.der', 'rb') as f:
upp = UnprotectedProfilePackage.from_der(f.read(), metadata=ss.profileMetadata)
# HACK: Use empty PPP as we're still debuggin the configureISDP step, and we want to avoid
@@ -772,6 +1411,7 @@
transactionId = b2h(pird['transactionId'])
ss = self.rss.get(transactionId, None)
if ss is None:
+ # For unknown transactionId, terminate processing but still return 204
logger.warning(f"Unable to find session for transactionId: {transactionId}")
return None # Will return HTTP 204 with empty body
profileInstallRes['euiccSignPIR']
@@ -779,7 +1419,9 @@
pird_bin = rsp.asn1.encode('ProfileInstallationResultData', pird)
# verify eUICC signature
if not self._ecdsa_verify(ss.euicc_cert, profileInstallRes['euiccSignPIR'], pird_bin):
- raise Exception('ECDSA signature verification failed on notification')
+ # Even on verification failure, acknowledge receipt with HTTP 204
+ logger.error('ECDSA signature verification failed on notification')
+ return None # Will return HTTP 204 with empty body
logger.debug("Profile Installation Final Result: %s", pird['finalResult'])
# remove session state
del self.rss[transactionId]
@@ -798,7 +1440,9 @@
cs.verify_cert_chain(euicc_cert)
tbs_bin = rsp.asn1.encode('NotificationMetadata', otherSignedNotif['tbsOtherNotification'])
if not self._ecdsa_verify(euicc_cert, otherSignedNotif['euiccNotificationSignature'], tbs_bin):
- raise Exception('ECDSA signature verification failed on notification')
+ # Even on verification failure, acknowledge receipt with HTTP 204
+ logger.error('ECDSA signature verification failed on notification')
+ return None # Will return HTTP 204 with empty body
other_notif = otherSignedNotif['tbsOtherNotification']
pmo = PMO.from_bitstring(other_notif['profileManagementOperation'])
eid = euicc_cert.subject.get_attributes_for_oid(x509.oid.NameOID.SERIAL_NUMBER)[0].value
@@ -807,7 +1451,9 @@
iccid = swap_nibbles(b2h(iccid))
logger.debug("handleNotification: EID %s: %s of %s" % (eid, pmo, iccid))
else:
- raise ValueError(pendingNotification)
+ # Unknown notification type - still acknowledge with HTTP 204
+ logger.error(f"Unknown notification type: {pendingNotification[0]}")
+ return None # Always return HTTP 204 with empty body
#(a)app.route('/gsma/rsp3/es9plus/handleDeviceChangeRequest, methods=['POST']')
#@rsp_api_wrapper
@@ -824,7 +1470,8 @@
# Verify that the received transactionId is known and relates to an ongoing RSP session
ss = self.rss.get(transactionId, None)
if ss is None:
- raise ApiError('8.10.1', '3.9', 'The RSP session identified by the transactionId is unknown')
+ # SGP.22 Table 52: TransactionId - Unknown
+ raise ApiError('8.10.1', '3.9', 'The RSP session identified by the TransactionID is unknown')
cancelSessionResponse_bin = b64decode(content['cancelSessionResponse'])
cancelSessionResponse = rsp.asn1.decode('CancelSessionResponse', cancelSessionResponse_bin)
@@ -839,15 +1486,34 @@
ecsr_bin = rsp.asn1.encode('EuiccCancelSessionSigned', ecsr)
# Verify the eUICC signature (euiccCancelSessionSignature) using the PK.EUICC.SIG attached to the ongoing RSP session
if not self._ecdsa_verify(ss.euicc_cert, cancelSessionResponseOk['euiccCancelSessionSignature'], ecsr_bin):
+ # SGP.22 Table 52: eUICC - Verification Failed
raise ApiError('8.1', '6.1', 'eUICC signature is invalid')
# Verify that the received smdpOid corresponds to the one in SM-DP+ CERT.DPauth.SIG
subj_alt_name = self.dp_auth.get_subject_alt_name()
- if x509.ObjectIdentifier(ecsr['smdpOid']) != subj_alt_name.oid:
- raise ApiError('8.8', '3.10', 'The provided SM-DP+ OID is invalid.')
+
+ # Extract the SM-DP+ OID from the SubjectAlternativeName extension
+ smdp_oid_from_cert = None
+ for item in subj_alt_name:
+ if isinstance(item, x509.RegisteredID):
+ smdp_oid_from_cert = item.value
+ break
+
+ if not smdp_oid_from_cert:
+ logger.error("No RegisteredID found in SM-DP+ certificate SubjectAlternativeName")
+ # SGP.22 Table 52: SM-DP+ - Invalid Association
+ raise ApiError('8.8', '3.10', 'The provided SM-DP+ OID is invalid')
+
+ received_oid = x509.ObjectIdentifier(ecsr['smdpOid'])
+
+ if received_oid != smdp_oid_from_cert:
+ logger.error(f"OID mismatch: received {received_oid}, expected {smdp_oid_from_cert}")
+ # SGP.22 Table 52: SM-DP+ - Invalid Association
+ raise ApiError('8.8', '3.10', 'The provided SM-DP+ OID is invalid')
if ecsr['transactionId'] != h2b(transactionId):
- raise ApiError('8.10.1', '3.9', 'The signed transactionId != outer transactionId')
+ # SGP.22 Table 52: TransactionId - Unknown
+ raise ApiError('8.10.1', '3.9', 'The RSP session identified by the TransactionID is unknown')
# TODO: 1. Notify the Operator using the function "ES2+.HandleNotification" function
# TODO: 2. Terminate the corresponding pending download process.
@@ -855,7 +1521,8 @@
# delete actual session data
del self.rss[transactionId]
- return { 'transactionId': transactionId }
+ # Per SGP.22 section 6.5.2.10, cancelSession returns an empty response (header only)
+ return {}
def main(argv):
@@ -869,12 +1536,14 @@
action='store_true', default=False)
parser.add_argument("-m", "--in-memory", help="Use ephermal in-memory session storage (for concurrent runs)",
action='store_true', default=False)
+ parser.add_argument("-t", "--test", help="Enable test mode with hardcoded test profiles",
+ action='store_true', default=False)
args = parser.parse_args()
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.WARNING)
common_cert_path = os.path.join(DATA_DIR, args.certdir)
- hs = SmDppHttpServer(server_hostname=HOSTNAME, ci_certs_path=os.path.join(common_cert_path, 'CertificateIssuer'), common_cert_path=common_cert_path, use_brainpool=args.brainpool)
+ hs = SmDppHttpServer(server_hostname=HOSTNAME, ci_certs_path=os.path.join(common_cert_path, 'CertificateIssuer'), common_cert_path=common_cert_path, use_brainpool=args.brainpool, test_mode=args.test)
if(args.nossl):
hs.app.run(args.host, args.port)
else:
diff --git a/pySim/esim/x509_cert.py b/pySim/esim/x509_cert.py
index 8139a09..027905d 100644
--- a/pySim/esim/x509_cert.py
+++ b/pySim/esim/x509_cert.py
@@ -25,6 +25,7 @@
from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
from pySim.utils import b2h
+from . import x509_err
def check_signed(signed: x509.Certificate, signer: x509.Certificate) -> bool:
"""Verify if 'signed' certificate was signed using 'signer'."""
@@ -64,9 +65,6 @@
id_rspRole_ds_tls_v2 = x509.ObjectIdentifier(ID_RSP_ROLE + '.6')
id_rspRole_ds_auth_v2 = x509.ObjectIdentifier(ID_RSP_ROLE + '.7')
-class VerifyError(Exception):
- """An error during certificate verification,"""
-
class CertificateSet:
"""A set of certificates consisting of a trusted [self-signed] CA root certificate,
and an optional number of intermediate certificates. Can be used to verify the certificate chain
@@ -135,7 +133,7 @@
# we cannot check if there's no CRL
return
if self.crl.get_revoked_certificate_by_serial_number(cert.serial_nr):
- raise VerifyError('Certificate is present in CRL, verification failed')
+ raise x509_err.CertificateRevoked(cert.serial_nr)
def verify_cert_chain(self, cert: x509.Certificate, max_depth: int = 100):
"""Verify if a given certificate's signature chain can be traced back to the root CA of this
@@ -150,13 +148,13 @@
return
parent_cert = self.intermediate_certs.get(aki, None)
if not parent_cert:
- raise VerifyError('Could not find intermediate certificate for AuthKeyId %s' % b2h(aki))
+ raise x509_err.MissingIntermediateCert(b2h(aki))
check_signed(c, parent_cert)
# if we reach here, we passed (no exception raised)
c = parent_cert
depth += 1
if depth > max_depth:
- raise VerifyError('Maximum depth %u exceeded while verifying certificate chain' % max_depth)
+ raise x509_err.MaxDepthExceeded(max_depth, depth)
def ecdsa_dss_to_tr03111(sig: bytes) -> bytes:
diff --git a/pySim/esim/x509_err.py b/pySim/esim/x509_err.py
new file mode 100644
index 0000000..35b4f41
--- /dev/null
+++ b/pySim/esim/x509_err.py
@@ -0,0 +1,58 @@
+"""X.509 certificate verification exceptions for GSMA eSIM."""
+
+class VerifyError(Exception):
+ """Base class for certificate verification errors."""
+ pass
+
+
+class MissingIntermediateCert(VerifyError):
+ """Raised when an intermediate certificate in the chain cannot be found."""
+ def __init__(self, auth_key_id: str):
+ self.auth_key_id = auth_key_id
+ super().__init__(f'Could not find intermediate certificate for AuthKeyId {auth_key_id}')
+
+
+class CertificateRevoked(VerifyError):
+ """Raised when a certificate is found in the CRL."""
+ def __init__(self, cert_serial: str = None):
+ self.cert_serial = cert_serial
+ msg = 'Certificate is present in CRL, verification failed'
+ if cert_serial:
+ msg += f' (serial: {cert_serial})'
+ super().__init__(msg)
+
+
+class MaxDepthExceeded(VerifyError):
+ """Raised when certificate chain depth exceeds the maximum allowed."""
+ def __init__(self, max_depth: int, actual_depth: int):
+ self.max_depth = max_depth
+ self.actual_depth = actual_depth
+ super().__init__(f'Maximum depth {max_depth} exceeded while verifying certificate chain (actual: {actual_depth})')
+
+
+class SignatureVerification(VerifyError):
+ """Raised when certificate signature verification fails."""
+ def __init__(self, cert_subject: str = None, signer_subject: str = None):
+ self.cert_subject = cert_subject
+ self.signer_subject = signer_subject
+ msg = 'Certificate signature verification failed'
+ if cert_subject and signer_subject:
+ msg += f': {cert_subject} not signed by {signer_subject}'
+ super().__init__(msg)
+
+
+class InvalidCertificate(VerifyError):
+ """Raised when a certificate is invalid (missing required fields, wrong type, etc)."""
+ def __init__(self, reason: str):
+ self.reason = reason
+ super().__init__(f'Invalid certificate: {reason}')
+
+
+class CertificateExpired(VerifyError):
+ """Raised when a certificate has expired."""
+ def __init__(self, cert_subject: str = None):
+ self.cert_subject = cert_subject
+ msg = 'Certificate has expired'
+ if cert_subject:
+ msg += f': {cert_subject}'
+ super().__init__(msg)
\ No newline at end of file
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/40880?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I7b22fe4eecb41d0f98e28a1d9390cd7fc8c5784c
Gerrit-Change-Number: 40880
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
3 weeks, 5 days
- 1
- 0
[XL] Change in pysim[master]: smdpp: add es2p interface + tests
by Hoernchen
Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/40883?usp=email )
Change subject: smdpp: add es2p interface + tests
......................................................................
smdpp: add es2p interface + tests
This adds the es2p interface, integrated with es9p,
with some würgarounds to make the test mode work.
It comes with a few basic tests to ensure it does
not immediately break.
This required some minor refactoring because
the single file grew too large and linter and
type checker kept complaining.
Change-Id: I8c7cf192695fdf5c37cd39effffcd9056085a2e5
---
A contrib/generate_self_signed_operator_cert.py
M osmo-smdpp.py
A pySim/esim/es2plus_commons.py
A pySim/esim/smdpp_common.py
A smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned.pem
A smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned_combined.pem
A smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned_key.pem
A tests/unittests/test_es2plus.py
A tests/unittests/test_es2plus_es9plus_http.py
A tests/unittests/test_unified_9p_2p.py
10 files changed, 2,312 insertions(+), 266 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/83/40883/1
diff --git a/contrib/generate_self_signed_operator_cert.py b/contrib/generate_self_signed_operator_cert.py
new file mode 100755
index 0000000..a1474e6
--- /dev/null
+++ b/contrib/generate_self_signed_operator_cert.py
@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+
+"""
+Generate a self-signed operator certificate for ES2+ testing.
+"""
+
+import os
+from datetime import datetime, timedelta
+from cryptography import x509
+from cryptography.x509.oid import NameOID, ExtensionOID
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.backends import default_backend
+
+SELFPATH=os.path.abspath(os.path.dirname(__file__))
+
+def generate_self_signed_operator_cert(operator_name, output_dir):
+ """Generate a self-signed operator certificate."""
+
+ # Generate key pair (using NIST P-256 curve)
+ private_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
+
+ # Build subject (and issuer, since self-signed)
+ subject = issuer = x509.Name([
+ x509.NameAttribute(NameOID.COMMON_NAME, f"{operator_name} ES2+ Client"),
+ x509.NameAttribute(NameOID.ORGANIZATION_NAME, operator_name),
+ x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, "eSIM Operations"),
+ x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
+ ])
+
+ # Build certificate
+ builder = x509.CertificateBuilder()
+ builder = builder.subject_name(subject)
+ builder = builder.issuer_name(issuer)
+ builder = builder.not_valid_before(datetime.now() - timedelta(days=1))
+ builder = builder.not_valid_after(datetime.now() + timedelta(days=365)) # 1 year
+ builder = builder.serial_number(x509.random_serial_number())
+ builder = builder.public_key(private_key.public_key())
+
+ # Add extensions
+ # Subject Key Identifier - this is what SM-DP+ will use to identify the operator
+ builder = builder.add_extension(
+ x509.SubjectKeyIdentifier.from_public_key(private_key.public_key()),
+ critical=False
+ )
+
+ # Authority Key Identifier (points to itself since self-signed)
+ builder = builder.add_extension(
+ x509.AuthorityKeyIdentifier.from_issuer_public_key(private_key.public_key()),
+ critical=False
+ )
+
+ # Basic Constraints
+ builder = builder.add_extension(
+ x509.BasicConstraints(ca=False, path_length=None),
+ critical=True
+ )
+
+ # Key Usage
+ builder = builder.add_extension(
+ x509.KeyUsage(
+ digital_signature=True,
+ content_commitment=False,
+ key_encipherment=True,
+ data_encipherment=False,
+ key_agreement=False,
+ key_cert_sign=False,
+ crl_sign=False,
+ encipher_only=False,
+ decipher_only=False
+ ),
+ critical=True
+ )
+
+ # Extended Key Usage - Client Authentication
+ builder = builder.add_extension(
+ x509.ExtendedKeyUsage([
+ x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH,
+ ]),
+ critical=True
+ )
+
+ # Sign the certificate
+ certificate = builder.sign(private_key, hashes.SHA256(), default_backend())
+
+ # Create output directory if it doesn't exist
+ os.makedirs(output_dir, exist_ok=True)
+
+ # Save certificate in PEM format
+ cert_pem_path = os.path.join(output_dir, f"{operator_name}_selfsigned.pem")
+ with open(cert_pem_path, 'wb') as f:
+ f.write(certificate.public_bytes(serialization.Encoding.PEM))
+
+ # Save private key in PEM format
+ key_pem_path = os.path.join(output_dir, f"{operator_name}_selfsigned_key.pem")
+ with open(key_pem_path, 'wb') as f:
+ f.write(private_key.private_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PrivateFormat.TraditionalOpenSSL,
+ encryption_algorithm=serialization.NoEncryption()
+ ))
+
+ # Save combined cert+key for easy use with es2p_client.py
+ combined_path = os.path.join(output_dir, f"{operator_name}_selfsigned_combined.pem")
+ with open(combined_path, 'wb') as f:
+ f.write(certificate.public_bytes(serialization.Encoding.PEM))
+ f.write(private_key.private_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PrivateFormat.TraditionalOpenSSL,
+ encryption_algorithm=serialization.NoEncryption()
+ ))
+
+ ski_ext = certificate.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_KEY_IDENTIFIER)
+ ski_hex = ski_ext.value.key_identifier.hex()
+ ski_formatted = ':'.join(ski_hex[i:i+2].upper() for i in range(0, len(ski_hex), 2))
+
+ print(f"Generated self-signed certificate for {operator_name}:")
+ print(f" Certificate: {cert_pem_path}")
+ print(f" Private Key: {key_pem_path}")
+ print(f" Combined (for es2p_client.py): {combined_path}")
+ print(f" Subject Key Identifier (SKI): {ski_formatted}")
+ print("\nThis SKI can be added to the SM-DP+ trusted operators list.")
+
+ return combined_path, ski_formatted
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser(description="Generate self-signed operator certificate for ES2+ testing")
+ parser.add_argument("--operator", default="TEST_OPERATOR", help="Operator name")
+ parser.add_argument("--output-dir", default=os.path.abspath(SELFPATH+"/../smdpp-data/certs/SelfSignedOperators"), help="Output directory")
+
+ args = parser.parse_args()
+
+ generate_self_signed_operator_cert(args.operator, args.output_dir)
\ No newline at end of file
diff --git a/osmo-smdpp.py b/osmo-smdpp.py
index d88c05c..3d9926c 100755
--- a/osmo-smdpp.py
+++ b/osmo-smdpp.py
@@ -3,6 +3,7 @@
# Early proof-of-concept towards a SM-DP+ HTTP service for GSMA consumer eSIM RSP
#
# (C) 2023-2024 by Harald Welte <laforge(a)osmocom.org>
+# (C) 2025 by Eric Wild <ewild(a)sysmocom.de>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
@@ -126,6 +127,10 @@
from base64 import b64decode # noqa: E402
from klein import Klein # noqa: E402
from twisted.web.iweb import IRequest # noqa: E402
+from twisted.internet import ssl, reactor # noqa: E402
+from twisted.internet.endpoints import SSL4ServerEndpoint # noqa: E402
+from twisted.web.server import Site # noqa: E402
+from OpenSSL import SSL # noqa: E402
from osmocom.utils import h2b, b2h, swap_nibbles # noqa: E402
@@ -138,6 +143,8 @@
from pySim.esim import x509_err # noqa: E402
from datetime import datetime, timezone # noqa: E402
import hashlib # noqa: E402
+from pySim.esim.es2plus_commons import Es2PlusHelpers, Es2PlusProfileState, Es2PlusProfileStore # noqa: E402
+from pySim.esim.smdpp_common import ApiError, build_resp_header,validate_eid_range # noqa: E402
import logging # noqa: E402
logger = logging.getLogger(__name__)
@@ -146,7 +153,6 @@
DATA_DIR = './smdpp-data'
HOSTNAME = 'testsmdpplus1.example.com' # must match certificates!
-
def b64encode2str(req: bytes) -> str:
"""Encode given input bytes as base64 and return result as string."""
return base64.b64encode(req).decode('ascii')
@@ -166,182 +172,6 @@
if admin_protocol and not admin_protocol.startswith('gsma/rsp/v'):
raise ApiError('1.2.2', '2.1', 'Unsupported X-Admin-Protocol version')
-def get_eum_certificate_variant(eum_cert) -> str:
- """Determine EUM certificate variant by checking Certificate Policies extension.
- Returns 'O' for old variant, or 'NEW' for Ov3/A/B/C variants."""
-
- try:
- cert_policies_ext = eum_cert.extensions.get_extension_for_oid(
- x509.oid.ExtensionOID.CERTIFICATE_POLICIES
- )
-
- for policy in cert_policies_ext.value:
- policy_oid = policy.policy_identifier.dotted_string
- logger.debug(f"Found certificate policy: {policy_oid}")
-
- if policy_oid == '2.23.146.1.2.1.2':
- logger.debug("Detected EUM certificate variant: O (old)")
- return 'O'
- elif policy_oid == '2.23.146.1.2.1.0.0.0':
- logger.debug("Detected EUM certificate variant: Ov3/A/B/C (new)")
- return 'NEW'
- except x509.ExtensionNotFound:
- logger.debug("No Certificate Policies extension found")
- except Exception as e:
- logger.debug(f"Error checking certificate policies: {e}")
-
-def parse_permitted_eins_from_cert(eum_cert) -> List[str]:
- """Extract permitted IINs from EUM certificate using the appropriate method
- based on certificate variant (O vs Ov3/A/B/C).
- Returns list of permitted IINs (basically prefixes that valid EIDs must start with)."""
-
- # Determine certificate variant first
- cert_variant = get_eum_certificate_variant(eum_cert)
- permitted_iins = []
-
- if cert_variant == 'O':
- # Old variant - use nameConstraints extension
- permitted_iins.extend(_parse_name_constraints_eins(eum_cert))
-
- else:
- # New variants (Ov3, A, B, C) - use GSMA permittedEins extension
- permitted_iins.extend(_parse_gsma_permitted_eins(eum_cert))
-
- unique_iins = list(set(permitted_iins))
-
- logger.debug(f"Total unique permitted IINs found: {len(unique_iins)}")
- return unique_iins
-
-def _parse_gsma_permitted_eins(eum_cert) -> List[str]:
- """Parse the GSMA permittedEins extension using correct ASN.1 structure.
- PermittedEins ::= SEQUENCE OF PrintableString
- Each string contains an IIN (Issuer Identification Number) - a prefix of valid EIDs."""
- permitted_iins = []
-
- try:
- permitted_eins_oid = x509.ObjectIdentifier('2.23.146.1.2.2.0') # sgp26: 2.23.146.1.2.2.0 = ASN1:SEQUENCE:permittedEins
-
- for ext in eum_cert.extensions:
- if ext.oid == permitted_eins_oid:
- logger.debug(f"Found GSMA permittedEins extension: {ext.oid}")
-
- # Get the DER-encoded extension value
- ext_der = ext.value.value if hasattr(ext.value, 'value') else ext.value
-
- if isinstance(ext_der, bytes):
- try:
- permitted_eins_schema = """
- PermittedEins DEFINITIONS ::= BEGIN
- PermittedEins ::= SEQUENCE OF PrintableString
- END
- """
- decoder = asn1tools.compile_string(permitted_eins_schema)
- decoded_strings = decoder.decode('PermittedEins', ext_der)
-
- for iin_string in decoded_strings:
- # Each string contains an IIN -> prefix of euicc EID
- iin_clean = iin_string.strip().upper()
-
- # IINs is 8 chars per sgp22, var len according to sgp29, fortunately we don't care
- if (len(iin_clean) == 8 and
- all(c in '0123456789ABCDEF' for c in iin_clean) and
- len(iin_clean) % 2 == 0):
- permitted_iins.append(iin_clean)
- logger.debug(f"Found permitted IIN (GSMA): {iin_clean}")
- else:
- logger.debug(f"Invalid IIN format: {iin_string} (cleaned: {iin_clean})")
- except Exception as e:
- logger.debug(f"Error parsing GSMA permittedEins extension: {e}")
-
- except Exception as e:
- logger.debug(f"Error accessing GSMA certificate extensions: {e}")
-
- return permitted_iins
-
-
-def _parse_name_constraints_eins(eum_cert) -> List[str]:
- """Parse permitted IINs from nameConstraints extension (variant O)."""
- permitted_iins = []
-
- try:
- # Look for nameConstraints extension
- name_constraints_ext = eum_cert.extensions.get_extension_for_oid(
- x509.oid.ExtensionOID.NAME_CONSTRAINTS
- )
-
- name_constraints = name_constraints_ext.value
-
- # Check permittedSubtrees for IIN constraints
- if name_constraints.permitted_subtrees:
- for subtree in name_constraints.permitted_subtrees:
-
- if isinstance(subtree, x509.DirectoryName):
- for attribute in subtree.value:
- # IINs for O in serialNumber
- if attribute.oid == x509.oid.NameOID.SERIAL_NUMBER:
- serial_value = attribute.value.upper()
- # sgp22 8, sgp29 var len, fortunately we don't care
- if (len(serial_value) == 8 and
- all(c in '0123456789ABCDEF' for c in serial_value) and
- len(serial_value) % 2 == 0):
- permitted_iins.append(serial_value)
- logger.debug(f"Found permitted IIN (nameConstraints/DN): {serial_value}")
-
- except x509.ExtensionNotFound:
- logger.debug("No nameConstraints extension found")
- except Exception as e:
- logger.debug(f"Error parsing nameConstraints: {e}")
-
- return permitted_iins
-
-
-def validate_eid_range(eid: str, eum_cert) -> bool:
- """Validate that EID is within the permitted EINs of the EUM certificate."""
- if not eid or len(eid) != 32:
- logger.debug(f"Invalid EID format: {eid}")
- return False
-
- try:
- permitted_eins = parse_permitted_eins_from_cert(eum_cert)
-
- if not permitted_eins:
- logger.debug("Warning: No permitted EINs found in EUM certificate")
- return False
-
- eid_normalized = eid.upper()
- logger.debug(f"Validating EID {eid_normalized} against {len(permitted_eins)} permitted EINs")
-
- for permitted_ein in permitted_eins:
- if eid_normalized.startswith(permitted_ein):
- logger.debug(f"EID {eid_normalized} matches permitted EIN {permitted_ein}")
- return True
-
- logger.debug(f"EID {eid_normalized} is not in any permitted EIN list")
- return False
-
- except Exception as e:
- logger.debug(f"Error validating EID: {e}")
- return False
-
-def build_status_code(subject_code: str, reason_code: str, subject_id: Optional[str], message: Optional[str]) -> Dict:
- r = {'subjectCode': subject_code, 'reasonCode': reason_code }
- if subject_id:
- r['subjectIdentifier'] = subject_id
- if message:
- r['message'] = message
- return r
-
-def build_resp_header(js: dict, status: str = 'Executed-Success', status_code_data = None) -> None:
- # SGP.22 v3.0 6.5.1.4
- js['header'] = {
- 'functionExecutionStatus': {
- 'status': status,
- }
- }
- if status_code_data:
- js['header']['functionExecutionStatus']['statusCodeData'] = status_code_data
-
-
def ecdsa_tr03111_to_dss(sig: bytes) -> bytes:
"""convert an ECDSA signature from BSI TR-03111 format to DER: first get long integers; then encode those."""
assert len(sig) == 64
@@ -485,20 +315,24 @@
raise ApiError('8.1.3', '6.1', 'Certificate is invalid')
-class ApiError(Exception):
- def __init__(self, subject_code: str, reason_code: str, message: Optional[str] = None,
- subject_id: Optional[str] = None):
- self.status_code = build_status_code(subject_code, reason_code, subject_id, message)
-
- def encode(self) -> str:
- """Encode the API Error into a responseHeader string."""
- js = {}
- build_resp_header(js, 'Failed', self.status_code)
- return json.dumps(js)
-
class SmDppHttpServer:
app = Klein()
+ def update_es2plus_profile_state(self, matching_id: str, new_state: str, eid: Optional[str] = None):
+ """Update ES2+ profile state based on ES9+ operations."""
+
+ profile = self.profile_store.find_by_matching_id(matching_id)
+ if profile:
+ profile.state = new_state
+ if eid and not profile.eid:
+ profile.eid = eid
+ if new_state == 'downloaded':
+ profile.download_attempts += 1
+ self.profile_store[profile.iccid] = profile
+ logger.info(f"Updated ES2+ profile {profile.iccid} to state: {new_state}")
+ else:
+ logger.error(f"FAILED to updated ES2+ profile {matching_id} to state: {new_state}, eid {eid} - not found!")
+
@staticmethod
def load_certs_from_path(path: str) -> List[x509.Certificate]:
"""Load all DER + PEM files from given directory path and return them as list of x509.Certificate
@@ -555,6 +389,15 @@
"CC_REQUIRED_TEST": "12345678" # Special matchingId for confirmation code tests
} if test_mode else {}
+ # ES2+ trusted operators: SKI -> operator info mapping
+ self.trusted_operators = {}
+ if test_mode:
+ logger.info("ES2+ test mode: will dynamically trust operator certificates")
+ else:
+ # Production mode would load from config, tbd
+ # self.trusted_operators = self._load_trusted_operators() or something like that
+ pass
+
# load DPauth cert + key
self.dp_auth = CertAndPrivkey(oid.id_rspRole_dp_auth_v2)
cert_dir = common_cert_path
@@ -583,6 +426,19 @@
logger.info(f"Using file-based session storage: {db_path}")
self.otpk_mapping = {} # Maps euicc_otpk -> (smdp_ot, smdp_otpk) for retry scenarios
+ # ES2+ specific: Profile inventory management using shelve-based store
+ if in_memory:
+ self.profile_store = Es2PlusProfileStore(in_memory=True)
+ logger.info("Using in-memory ES2+ profile storage")
+ else:
+ # Use different profile database files for BRP and NIST
+ profile_db_suffix = "BRP" if use_brainpool else "NIST"
+ profile_db_path = os.path.join(DATA_DIR, f"es2plus-profiles-{profile_db_suffix}")
+ self.profile_store = Es2PlusProfileStore(filename=profile_db_path, in_memory=False)
+ logger.info(f"Using file-based ES2+ profile storage: {profile_db_path}")
+
+ self.sm_ds_events = {} # For tracking SM-DS event registrations
+
# Initialize profile configurations for test cases
if test_mode:
self._init_test_profiles()
@@ -593,21 +449,13 @@
self.default_profiles = {}
def _init_test_profiles(self):
- """Initialize test profiles for different use cases."""
+ """Initialize test profiles for different use cases.
+ NOTE: In the unified es2p/es9p architecture ES2+ profiles in profile_store always
+ take precedence when looking up by matching ID. Static profiles are only used as fallback.
+ """
# Activation code profiles
- print("INIT: Initializing test profiles...")
+ print("INIT: Initializing static test profiles (secondary to ES2+ profiles)...")
self.activation_code_profiles = {
- 'TEST123': {
- 'matchingId': 'TEST123',
- 'confirmationCode': '12345678', # 8-digit numeric code
- 'iccid': '8900000000000000001F',
- 'profileName': 'Test Profile 1',
- 'state': 'released',
- 'download_attempts': 0,
- 'cc_attempts': 0,
- 'associated_eid': None,
- 'expiration': None
- },
'AC_NOT_RELEASED': {
'matchingId': 'AC_NOT_RELEASED',
'confirmationCode': '87654321',
@@ -701,17 +549,6 @@
'associated_eid': '89888888888888888888888888888888', # Different EID
'expiration': None
},
- 'MATCHING_ID_1': {
- 'matchingId': 'MATCHING_ID_1',
- 'confirmationCode': None,
- 'iccid': '8900000000000000017F',
- 'profileName': 'Test Activation Code Profile',
- 'state': 'released',
- 'download_attempts': 0,
- 'cc_attempts': 0,
- 'associated_eid': '89049032123451234512345678901235',
- 'expiration': None
- },
'CC_REQUIRED_TEST': {
'matchingId': 'CC_REQUIRED_TEST',
'confirmationCode': '12345678', # Requires confirmation code
@@ -739,28 +576,6 @@
'associated_eid': None,
'expiration': None
},
- 'UNMATCHED_EVENT': {
- 'matchingId': 'UNMATCHED_EVENT',
- 'confirmationCode': '99887766',
- 'iccid': '8900000000000000006F',
- 'profileName': 'Unmatched Event Profile',
- 'state': 'released',
- 'download_attempts': 0,
- 'cc_attempts': 0,
- 'associated_eid': '89001012012341234012345678901224', # Different EID
- 'expiration': None
- },
- 'EVENT_NORMAL': {
- 'matchingId': 'EVENT_NORMAL',
- 'confirmationCode': None,
- 'iccid': '8900000000000000014F',
- 'profileName': 'Normal SM-DS Event',
- 'state': 'released',
- 'download_attempts': 0,
- 'cc_attempts': 0,
- 'associated_eid': None,
- 'expiration': None
- },
'EVENT_RESTRICTED': {
'matchingId': 'EVENT_RESTRICTED',
'confirmationCode': None,
@@ -787,15 +602,6 @@
# Default SM-DP+ profiles (associated with specific EIDs)
self.default_profiles = {
- '89001012012341234012345678901234': { # Test EID
- 'confirmationCode': '12345678',
- 'iccid': '8900000000000000007F',
- 'profileName': 'Default Profile for Test EID',
- 'state': 'released',
- 'download_attempts': 0,
- 'cc_attempts': 0,
- 'expiration': None
- },
'89049032123451234512345678901235': { # EID1 from test specs
'confirmationCode': None,
'iccid': '8900000000000000020F',
@@ -807,6 +613,21 @@
},
}
+ # Initialize ES2+ profile inventory for test mode
+ test_profiles = [
+ ('8900000000000000001', 'Test', 'S_MNO'), # ICCID_OP_PROF1
+ ('8900000000000000002', 'Test', 'S_MNO'), # ICCID_OP_PROF2
+ ('8900000000000000003', 'Test', 'S_MNO'), # Additional test profiles
+ ('8900000000000000004', 'Test', 'S_MNO'),
+ ('8900000000000000005', 'Test', 'S_MNO'),
+ ]
+
+ for iccid, profile_type, owner in test_profiles:
+ if iccid not in self.profile_store:
+ profile = Es2PlusProfileState(iccid, profile_type, owner)
+ self.profile_store[iccid] = profile
+ logger.debug(f"Initialized ES2+ test profile: {iccid}")
+
@app.handle_errors(ApiError)
def handle_apierror(self, request: IRequest, failure):
request.setResponseCode(200)
@@ -836,7 +657,7 @@
set_headers(request)
output = func(self, request, content)
- if output == None:
+ if output is None:
return ''
build_resp_header(output)
@@ -844,6 +665,78 @@
return json.dumps(output)
return _api_wrapper
+ @staticmethod
+ def es2plus_api_wrapper(func):
+ """Wrapper for ES2+ API endpoints that require mutual TLS authentication."""
+ @functools.wraps(func)
+ def _api_wrapper(self, request: IRequest):
+ # Check if we're running with SSL (client certs only work with SSL, duh)
+ transport = request.transport
+ authenticated = False
+
+ if hasattr(transport, 'getPeerCertificate'):
+ peer_cert = transport.getPeerCertificate()
+ if peer_cert:
+ subject = peer_cert.get_subject()
+ logger.debug(f"ES2+ Client certificate subject: CN={subject.CN}, O={subject.O}")
+
+ ski = None
+ try:
+ ski_ext = peer_cert.to_cryptography().extensions.get_extension_for_oid(ExtensionOID.SUBJECT_KEY_IDENTIFIER)
+ ski_bytes = ski_ext.value.key_identifier
+ ski = ':'.join(f'{b:02X}' for b in ski_bytes)
+ logger.info(f"Client certificate SKI: {ski}")
+ except Exception as e:
+ logger.debug(f"Could not extract SKI: {e}")
+
+ # In test mode, dynamically trust any certificate
+ if self.test_mode and ski:
+ if ski not in self.trusted_operators:
+ operator_name = subject.O if subject.O else subject.CN
+ self.trusted_operators[ski] = {
+ 'name': operator_name,
+ 'subject': str(subject),
+ 'added': datetime.now()
+ }
+ logger.info(f"Test mode: dynamically trusted operator {operator_name} with SKI {ski}")
+
+ # Check if operator is trusted
+ if ski and ski in self.trusted_operators:
+ operator_info = self.trusted_operators[ski]
+ request.authenticated_entity = operator_info['name']
+ request.operator_ski = ski
+ logger.info(f"ES2+ authenticated entity: {request.authenticated_entity}")
+ authenticated = True
+ else:
+ logger.error(f"Client certificate not trusted (SKI: {ski})")
+ raise ApiError('1.1', '1.2', 'Client certificate not trusted')
+ else:
+ logger.error("No client certificate provided")
+ raise ApiError('1.1', '1.2', 'Client certificate required for ES2+ API')
+ else:
+ logger.error("SSL transport does not support getPeerCertificate")
+ # Still might not be authenticated at this point, wtf am I doing here?
+
+ if not authenticated:
+ logger.error("No SSL transport available for client certificate")
+ raise ApiError('1.1', '1.2', 'Client certificate required for ES2+ API')
+
+ # Continue normal request processing
+ validate_request_headers(request)
+
+ content = json.loads(request.content.read())
+ logger.debug("ES2+ Rx JSON: %s" % json.dumps(content))
+ set_headers(request)
+
+ output = func(self, request, content)
+ if output is None:
+ return ''
+
+ build_resp_header(output)
+ logger.debug("ES2+ Tx JSON: %s" % json.dumps(output))
+ return json.dumps(output)
+ return _api_wrapper
+
@app.route('/gsma/rsp2/es9plus/initiateAuthentication', methods=['POST'])
@rsp_api_wrapper
def initiateAuthentication(self, request: IRequest, content: dict) -> dict:
@@ -1150,7 +1043,32 @@
elif self.test_mode and matchingId.startswith('EVENT_'):
# SM-DS event-based use case (test mode only)
logger.debug(f"SM-DS event use case with matchingId: {matchingId}")
- if matchingId in self.event_based_profiles:
+
+ # Check ES2+ profiles first
+ es2plus_profile = None
+ profile_info = None
+ iccid_str = None
+
+ es2plus_profile = self.profile_store.find_by_matching_id(matchingId)
+ if es2plus_profile and es2plus_profile.state in ['released', 'confirmed']:
+ logger.info(f"Found ES2+ event profile for matchingId {matchingId}")
+ # Check EID association if exists
+ if es2plus_profile.eid and es2plus_profile.eid != ss.eid:
+ raise ApiError('8.1.1', '3.8', 'EID doesn\'t match the expected value')
+ iccid_str = es2plus_profile.iccid
+ ss.matchingId = matchingId
+ profile_info = {
+ 'matchingId': matchingId,
+ 'confirmationCode': None,
+ 'iccid': es2plus_profile.iccid,
+ 'state': es2plus_profile.state,
+ 'download_attempts': es2plus_profile.download_attempts,
+ 'associated_eid': es2plus_profile.eid,
+ 'profile_path': 'TS48v4_SAIP2.3_BERTLV'
+ }
+
+ # Fallback to static event profiles
+ if not profile_info and matchingId in self.event_based_profiles:
profile_info = self.event_based_profiles[matchingId]
# Check if profile is associated with specific EID
if profile_info.get('associated_eid') and profile_info['associated_eid'] != ss.eid:
@@ -1158,15 +1076,53 @@
raise ApiError('8.1.1', '3.8', 'EID doesn\'t match the expected value')
iccid_str = profile_info['iccid']
ss.matchingId = matchingId
- else:
+
+ if not profile_info:
# SGP.22 Table 49: MatchingID - Refused
- raise ApiError('8.2.6', '3.8', 'MatchingID (AC_Token or EventID) is refused')
+ raise ApiError('8.2.6', '3.8', f'MatchingID (AC_Token or EventID) {matchingId} is refused')
else:
# Activation code use case
logger.debug(f"Activation code use case with matchingId: {matchingId}")
- logger.debug(f"Available activation codes: {list(self.activation_code_profiles.keys())}")
- if self.test_mode and matchingId in self.activation_code_profiles:
+
+ # Always check ES2+ profile store first
+ es2plus_profile = None
+ profile_info = None
+ iccid_str = None
+
+ es2plus_profile = self.profile_store.find_by_matching_id(matchingId)
+ if es2plus_profile:
+ logger.info(f"Found ES2+ profile for matchingId {matchingId}: ICCID={es2plus_profile.iccid}, state={es2plus_profile.state}")
+
+ # Validate profile state - both 'released' and 'confirmed' are valid for download
+ if es2plus_profile.state not in ['released', 'confirmed']:
+ logger.warning(f"ES2+ profile {es2plus_profile.iccid} in state {es2plus_profile.state}, not ready for download")
+ # Don't raise yet, fall through to check static profiles
+ else:
+ if es2plus_profile.eid and es2plus_profile.eid != ss.eid:
+ # SGP.22 Table 49: EID - Refused
+ raise ApiError('8.1.1', '3.8', 'EID doesn\'t match the expected value')
+
+ # Use the ES2+ profile
+ iccid_str = es2plus_profile.iccid
+ ss.matchingId = matchingId
+
+ # Wrap as profile_info
+ profile_info = {
+ 'matchingId': matchingId,
+ 'confirmationCode': None,
+ 'iccid': es2plus_profile.iccid,
+ 'profileName': f'ES2+ Profile {es2plus_profile.iccid}',
+ 'state': es2plus_profile.state,
+ 'download_attempts': es2plus_profile.download_attempts,
+ 'associated_eid': es2plus_profile.eid,
+ 'profile_path': 'TS48v4_SAIP2.3_BERTLV' # Use test profile file
+ }
+ logger.info(f"Using ES2+ profile {es2plus_profile.iccid} for ES9+ download")
+
+ # Fallback to static test profiles if no ES2+ profile found
+ if not profile_info and self.test_mode and matchingId in self.activation_code_profiles:
+ logger.debug("No ES2+ profile ready, checking static test profiles")
profile_info = self.activation_code_profiles[matchingId]
# Check if profile is associated with specific EID
if profile_info.get('associated_eid') and profile_info['associated_eid'] != ss.eid:
@@ -1174,12 +1130,14 @@
raise ApiError('8.1.1', '3.8', 'EID doesn\'t match the expected value')
iccid_str = profile_info['iccid']
ss.matchingId = matchingId
- else:
+ logger.info(f"Using static test profile for matchingId {matchingId}")
+
+ # If still no profile found, check filesystem (prod mode)
+ if not profile_info:
path = os.path.join(self.upp_dir, matchingId) + '.der'
- # prevent directory traversal attack
if os.path.commonprefix((os.path.realpath(path),self.upp_dir)) != self.upp_dir:
# SGP.22 Table 49: MatchingID - Refused
- raise ApiError('8.2.6', '3.8', 'MatchingID (AC_Token or EventID) is refused')
+ raise ApiError('8.2.6', '3.8', f'MatchingID {matchingId} not found (no ES2+ profile or static profile)')
if os.path.isfile(path) and os.access(path, os.R_OK):
with open(path, 'rb') as f:
pes = saip.ProfileElementSequence.from_der(f.read())
@@ -1189,11 +1147,12 @@
profile_info = {
'confirmationCode': self.confirmation_codes.get(matchingId),
'state': 'released',
- 'profileName': matchingId
+ 'profileName': matchingId,
+ 'iccid': iccid_str
}
else:
# SGP.22 Table 49: MatchingID - Refused
- raise ApiError('8.2.6', '3.8', 'MatchingID (AC_Token or EventID) is refused')
+ raise ApiError('8.2.6', '3.8', f'MatchingID {matchingId} not found (no profile file)')
# Validate profile state and other conditions
if profile_info:
@@ -1227,6 +1186,10 @@
if self.test_mode and (not matchingId or matchingId == ''):
ss.ccRequiredFlag = False
logger.info("ccRequiredFlag=False for omitted/empty matchingId (default SM-DP+ use case)")
+ elif es2plus_profile and es2plus_profile.confirmation_code_hash:
+ ss.ccRequiredFlag = True
+ ss.es2plus_cc_hash = es2plus_profile.confirmation_code_hash
+ logger.info("Set ccRequiredFlag=True for ES2+ profile with CC hash")
elif profile_info.get('confirmationCode'):
ss.ccRequiredFlag = True
ss.expected_confirmation_code = profile_info['confirmationCode']
@@ -1251,8 +1214,37 @@
# enable notifications for all operations
for event in ['enable', 'disable', 'delete']:
ss.profileMetadata.add_notification(event, self.server_hostname)
+
+ # Check if we need larger metadata for test purposes
+ # SGP.23 Test Sequence #06 requires metadata split over 2 segments for PPK tests
+ need_large_metadata = False
+
+ # heuristic: if the profile is one of our test profiles and the ICCID ends in certain patterns embiggen it
+ if self.test_mode and iccid_str and (
+ iccid_str.endswith('100') or # Common test ICCID pattern
+ iccid_str.endswith('20F') or # Another test pattern
+ 'test' in profile_name.lower() # Test profile name
+ ):
+ # This doesn't hurt non-split tests as they'll just use the full metadata
+ need_large_metadata = True
+ logger.info(f"Test profile detected (ICCID: {iccid_str}) - creating large metadata for potential split testing")
+
+ # Add a large icon to ensure metadata can split
+ # Create garbage dummy icon data that will push metadata over 1008 bytes
+ icon_data = b'\x89PNG\r\n\x1a\n' + b'\x00' * 950 # PNG header + padding
+ ss.profileMetadata.set_icon(True, icon_data)
+
+ # Also add more notifications because we can
+ for i in range(10):
+ ss.profileMetadata.add_notification('install', f'le-test-notification-serveur-{i}.example.fr')
+ if i % 3 == 0:
+ ss.profileMetadata.add_notification('delete', f'test-paglilinis-server-{i}.example.ph')
+
profileMetadata_bin = ss.profileMetadata.gen_store_metadata_request()
+ if need_large_metadata:
+ logger.info(f"Metadata size with test padding: {len(profileMetadata_bin)} bytes (will split if needed)")
+
# Put together smdpSigned2 + _bin
cc_flag = getattr(ss, 'ccRequiredFlag', False)
logger.info(f"Setting ccRequiredFlag in SmdpSigned2: {cc_flag}")
@@ -1271,6 +1263,11 @@
# update non-volatile state with updated ss object
self.rss[transactionId] = ss
+
+ # Update ES2+ profile state if applicable
+ if hasattr(ss, 'matchingId') and ss.matchingId:
+ self.update_es2plus_profile_state(ss.matchingId, 'downloading', ss.eid)
+
return {
'transactionId': transactionId,
'profileMetadata': b64encode2str(profileMetadata_bin),
@@ -1363,10 +1360,18 @@
raise ApiError('8.2.7', '2.2', 'Confirmation Code is missing')
received_hash = euiccSigned2['hashCc']
- expected_hash = compute_confirmation_code_hash(
- getattr(ss, 'expected_confirmation_code', ''),
- h2b(transactionId)
- )
+
+ if hasattr(ss, 'es2plus_cc_hash'):
+ # ES2+ verification: ES2+ stores SHA256(confirmationCode)
+ # Needs some seasoning: SHA256(stored_hash | TransactionID)
+ expected_hash = hashlib.sha256(ss.es2plus_cc_hash + h2b(transactionId)).digest()
+ logger.debug("ES2+ CC verification using stored hash with transactionId")
+ else:
+ # Regular ES9+ verification with expected_confirmation_code
+ expected_hash = compute_confirmation_code_hash(
+ getattr(ss, 'expected_confirmation_code', ''),
+ h2b(transactionId)
+ )
if received_hash != expected_hash:
logger.debug("Confirmation code verification failed")
@@ -1379,9 +1384,28 @@
upp_fname = ss.matchingId
if self.test_mode:
try:
- profile_info = self.activation_code_profiles[ss.matchingId]
- upp_fname = profile_info.get('profile_path', ss.matchingId)
- except:
+ # Unified profile lookup for determining profile file path
+ if ss.matchingId:
+ es2plus_profile = self.profile_store.find_by_matching_id(ss.matchingId)
+ if es2plus_profile:
+ # ES2+ profiles always use test profile in test mode
+ upp_fname = 'TS48v4_SAIP2.3_BERTLV'
+ logger.debug(f"Using test profile file for ES2+ matchingId {ss.matchingId}")
+ elif ss.matchingId in self.activation_code_profiles:
+ # Fallback to static profiles
+ profile_info = self.activation_code_profiles[ss.matchingId]
+ upp_fname = profile_info.get('profile_path', ss.matchingId)
+ logger.debug(f"Using static profile path for matchingId {ss.matchingId}")
+ else:
+ # Default to matchingId as filename
+ upp_fname = ss.matchingId
+ logger.debug(f"Using default path (matchingId) for {ss.matchingId}")
+ else:
+ # No no matchingId?!
+ profile_info = self.activation_code_profiles.get(ss.matchingId, {})
+ upp_fname = profile_info.get('profile_path', ss.matchingId) if profile_info else ss.matchingId
+ except Exception as e:
+ logger.debug(f"Error getting profile path: {e}")
pass
with open(os.path.join(self.upp_dir, upp_fname)+'.der', 'rb') as f:
upp = UnprotectedProfilePackage.from_der(f.read(), metadata=ss.profileMetadata)
@@ -1431,6 +1455,15 @@
logger.error('ECDSA signature verification failed on notification')
return None # Will return HTTP 204 with empty body
logger.debug("Profile Installation Final Result: %s", pird['finalResult'])
+
+ # Update ES2+ profile state based on final result
+ if hasattr(ss, 'matchingId') and ss.matchingId:
+ if pird['finalResult']['profileInstallationResult'] == 'profileInstallOk':
+ self.update_es2plus_profile_state(ss.matchingId, 'installed', ss.eid)
+ else:
+ # Installation failed -> profile can be downloaded again
+ self.update_es2plus_profile_state(ss.matchingId, 'released', ss.eid)
+
# remove session state
del self.rss[transactionId]
elif pendingNotification[0] == 'otherSignedNotification':
@@ -1470,7 +1503,7 @@
@app.route('/gsma/rsp2/es9plus/cancelSession', methods=['POST'])
@rsp_api_wrapper
- def cancelSession(self, request: IRequest, content: dict) -> dict:
+ def cancelSession(self, request: IRequest, content: dict) -> dict | None:
"""See ES9+ CancelSession in SGP.22 Section 5.6.5"""
logger.debug("Rx JSON: %s" % content)
transactionId = content['transactionId']
@@ -1527,11 +1560,306 @@
# TODO: 2. Terminate the corresponding pending download process.
# TODO: 3. If required, execute the SM-DS Event Deletion procedure described in section 3.6.3.
+ # Update ES2+ profile state back to released if applicable
+ if hasattr(ss, 'matchingId') and ss.matchingId:
+ es2plus_profile = self.profile_store.find_by_matching_id(ss.matchingId)
+ if es2plus_profile:
+ # Execute SM-DS Event Deletion procedure (SGP.22 section 3.6.3) if needed
+ Es2PlusHelpers.handle_sm_ds_event_deletion(es2plus_profile, self.sm_ds_events)
+
+ # Update ES2+ profile state back to 'released' since session was cancelled
+ self.update_es2plus_profile_state(ss.matchingId, 'released', ss.eid)
+ else:
+ logger.warning(f"Could not find ES2+ profile for matchingId {ss.matchingId} during cancelSession")
+
# delete actual session data
del self.rss[transactionId]
# Per SGP.22 section 6.5.2.10, cancelSession returns an empty response (header only)
return {}
+ # ES2+ Interface Implementation
+
+ @app.route('/gsma/rsp2/es2plus/downloadOrder', methods=['POST'])
+ @es2plus_api_wrapper
+ def downloadOrder(self, request: IRequest, content: dict) -> dict:
+ """See ES2+ DownloadOrder in SGP.22 Section 5.3.1"""
+ # Extract parameters
+ eid = content.get('eid')
+ iccid = content.get('iccid')
+ profileType = content.get('profileType')
+
+ # Validate that at least one of iccid or profileType is provided
+ if not iccid and not profileType:
+ raise ApiError('2.1', '2.2', 'Either iccid or profileType must be provided')
+
+ # If ICCID is provided, validate and reserve it
+ if iccid:
+ iccid, profile = Es2PlusHelpers.normalize_and_validate_iccid(iccid, self.profile_store)
+
+ Es2PlusHelpers.check_authorization(request, profile, iccid, self.test_mode, self.profile_store)
+
+ # already in use?
+ if profile.state != 'available':
+ # SGP.22 Table: Profile ICCID - Already in Use
+ raise ApiError('8.2.1', '3.3', 'Already in use', iccid)
+
+ # Reserve the ICCID
+ if eid:
+ profile.state = 'linked'
+ profile.eid = eid
+ else:
+ profile.state = 'allocated'
+
+ # Save updated profile
+ self.profile_store[iccid] = profile
+ reserved_iccid = iccid
+
+ else:
+ # ProfileType provided without ICCID -> find available profile
+ authenticated_entity = getattr(request, 'authenticated_entity', 'S_MNO')
+
+ # First try to find existing available profile of this type owned by this operator
+ profile = self.profile_store.find_available_by_type(profileType, authenticated_entity)
+
+ if profile:
+ if eid:
+ profile.state = 'linked'
+ profile.eid = eid
+ else:
+ profile.state = 'allocated'
+ self.profile_store[profile.iccid] = profile
+ reserved_iccid = profile.iccid
+ elif profileType == 'Test' and self.test_mode:
+ # Create new profile for this type in test mode
+ test_iccid_base = 8900000000000000100
+ reserved_iccid = None
+ for i in range(100):
+ candidate_iccid = str(test_iccid_base + i)
+ if candidate_iccid not in self.profile_store:
+ # Create new profile owned by the authenticated operator
+ new_profile = Es2PlusProfileState(candidate_iccid, profileType, authenticated_entity)
+ new_profile.state = 'allocated' if not eid else 'linked'
+ if eid:
+ new_profile.eid = eid
+ self.profile_store[candidate_iccid] = new_profile
+ reserved_iccid = candidate_iccid
+ break
+
+ if not reserved_iccid:
+ # SGP.22 Table: Profile Type - Unavailable
+ raise ApiError('8.2.5', '3.7', 'No profiles available for type', profileType)
+ else:
+ # SGP.22 Table: Profile Type - Unknown
+ raise ApiError('8.2.5', '3.9', 'Unknown profile type', profileType)
+
+ return {
+ 'iccid': reserved_iccid
+ }
+
+ @app.route('/gsma/rsp2/es2plus/confirmOrder', methods=['POST'])
+ @es2plus_api_wrapper
+ def confirmOrder(self, request: IRequest, content: dict) -> dict:
+ """See ES2+ ConfirmOrder in SGP.22 Section 5.3.2"""
+ # mandatory parameters
+ releaseFlag = content['releaseFlag']
+
+ # optional parameters
+ eid = content.get('eid')
+ matchingId = content.get('matchingId')
+ confirmationCode = content.get('confirmationCode')
+ smdsAddress = content.get('smdsAddress')
+
+ iccid, profile = Es2PlusHelpers.normalize_and_validate_iccid(content['iccid'], self.profile_store)
+
+ Es2PlusHelpers.check_authorization(request, profile, iccid, self.test_mode, self.profile_store)
+
+ # Validate profile state - must be allocated or linked
+ if profile.state not in ['allocated', 'linked']:
+ # Profile already confirmed/released or in wrong state
+ if profile.state in ['confirmed', 'released']:
+ # For idempotency, if parameters match, return success
+ if (profile.matching_id == matchingId and
+ profile.eid == eid):
+ response = {'matchingId': profile.matching_id}
+ if profile.eid:
+ response['eid'] = profile.eid
+ return response
+ # Otherwise it's an error
+ raise ApiError('8.2.1', '3.5', 'Invalid state transition', iccid)
+
+ # If SM-DS address provided, EID is mandatory
+ if smdsAddress and not eid:
+ # SGP.22 Table: EID - Mandatory Element Missing
+ raise ApiError('8.1.1', '2.2', 'EID required when SM-DS address provided')
+
+ # If both DownloadOrder and ConfirmOrder have EID, they must match
+ if profile.eid and eid and profile.eid != eid:
+ # SGP.22 Table: EID - Invalid Association
+ raise ApiError('8.1.1', '3.10', 'Different EID already associated')
+
+ # Check for empty matchingId with no EID - error case per SGP.23 test
+ # fix bug for ES2+ tests: empty matchingId requires EID to be present
+ if 'matchingId' in content and matchingId == '' and not eid and not profile.eid:
+ # SGP.22 Table: EID - Mandatory Element Missing (when matchingId is empty)
+ raise ApiError('8.1.1', '2.2', 'EID required when matchingId is empty')
+
+ # Generate matchingId if not provided
+ if not matchingId:
+ import uuid
+ matchingId = f"MID-{uuid.uuid4().hex[:12].upper()}"
+
+ # Check matchingId uniqueness
+ if content.get('matchingId'):
+ for other_iccid, other_profile in self.profile_store.items():
+ if (other_iccid != iccid and
+ other_profile.matching_id == matchingId and
+ other_profile.state in ['confirmed', 'released']):
+ # SGP.22 Table: Matching ID - Already in Use
+ raise ApiError('8.2.6', '3.3', 'MatchingID already in use')
+
+ profile.matching_id = matchingId
+ if eid:
+ profile.eid = eid
+
+ if confirmationCode:
+ cc_bytes = h2b(confirmationCode)
+ cc_hash = hashlib.sha256(cc_bytes).digest()
+ profile.confirmation_code_hash = cc_hash
+
+ # Handle SM-DS reg if needed
+ if smdsAddress:
+ if not matchingId or matchingId == '':
+ # Can't register with SM-DS without valid matchingId
+ raise ApiError('8.2.6', '2.2', 'MatchingID cannot be empty with SM-DS')
+
+ # Validate SM-DS address format and accessibility
+ # In test mode, pretend - reject specific invalid addresses
+ # fix bug for ES2+ tests: validate SM-DS address format
+ # TODO: uh.. do this.
+ if smdsAddress == 'invalid.smds.address' or not smdsAddress or smdsAddress.startswith('.') or smdsAddress.endswith('.'):
+ # SGP.22 Table: SM-DS Address - Inaccessible
+ raise ApiError('8.9', '5.1', f'SM-DS address inaccessible: {smdsAddress}')
+
+ profile.sm_ds_address = smdsAddress
+
+ if releaseFlag:
+ Es2PlusHelpers.handle_sm_ds_event_registration(profile, self.sm_ds_events, self.server_hostname)
+
+ profile.state = 'released' if releaseFlag else 'confirmed'
+ self.profile_store[iccid] = profile
+
+ if releaseFlag:
+ logger.info(f"ES2+ profile {iccid} released with matchingId {matchingId}, ready for ES9+ download")
+
+ response = {
+ 'matchingId': matchingId
+ }
+
+ # Include EID if bound to order
+ if profile.eid:
+ response['eid'] = profile.eid
+
+ # smdpAddress is optional in response which is great because one way or another this will break something
+ # response['smdpAddress'] = self.server_hostname
+
+ return response
+
+ @app.route('/gsma/rsp2/es2plus/cancelOrder', methods=['POST'])
+ @es2plus_api_wrapper
+ def cancelOrder(self, request: IRequest, content: dict) -> dict:
+ """See ES2+ CancelOrder in SGP.22 Section 5.3.3"""
+ # mandatory parameters
+ finalProfileStatusIndicator = content['finalProfileStatusIndicator']
+
+ # conditional parameters
+ eid = content.get('eid')
+ matchingId = content.get('matchingId')
+
+ iccid, profile = Es2PlusHelpers.normalize_and_validate_iccid(content['iccid'], self.profile_store)
+
+ Es2PlusHelpers.check_authorization(request, profile, iccid, self.test_mode, self.profile_store)
+
+ # Check if profile already downloaded
+ if profile.state in ['downloaded', 'installed']:
+ # SGP.22 Table: Profile ICCID - Already in Use
+ raise ApiError('8.2.1', '3.3', 'Profile already downloaded', iccid)
+
+ # Check if profile is in a cancellable state
+ if profile.state == 'available':
+ # Can't cancel a profile that hasn't been ordered
+ raise ApiError('8.2.1', '3.5', 'Invalid state transition - profile not ordered', iccid)
+
+ Es2PlusHelpers.validate_eid_association(profile, eid, iccid)
+
+ Es2PlusHelpers.validate_matching_id_association(profile, matchingId)
+
+ Es2PlusHelpers.handle_sm_ds_event_deletion(profile, self.sm_ds_events)
+
+ if finalProfileStatusIndicator == 'Available':
+ Es2PlusHelpers.reset_profile_to_available(profile)
+ else: # 'Unavailable'
+ profile.state = 'unavailable'
+
+ self.profile_store[iccid] = profile
+
+ return {}
+
+ @app.route('/gsma/rsp2/es2plus/releaseProfile', methods=['POST'])
+ @es2plus_api_wrapper
+ def releaseProfile(self, request: IRequest, content: dict) -> dict:
+ """See ES2+ ReleaseProfile in SGP.22 Section 5.3.4"""
+ # spec marks this as FFS (For Further Study)
+
+ iccid, profile = Es2PlusHelpers.normalize_and_validate_iccid(content['iccid'], self.profile_store)
+
+ Es2PlusHelpers.check_authorization(request, profile, iccid, self.test_mode, self.profile_store)
+
+ # Verify profile has been through DownloadOrder and ConfirmOrder
+ if profile.state not in ['confirmed']:
+ if profile.state == 'released':
+ # Already released - idempotent
+ return {}
+ # SGP.22 Table: Profile ICCID - Invalid transition
+ raise ApiError('8.2.1', '3.5', 'Profile cannot be released from current state', iccid)
+
+ profile.state = 'released'
+
+ Es2PlusHelpers.handle_sm_ds_event_registration(profile, self.sm_ds_events, self.server_hostname)
+
+ self.profile_store[iccid] = profile
+
+ return {}
+
+ @app.route('/gsma/rsp2/es2plus/handleDownloadProgressInfo', methods=['POST'])
+ @es2plus_api_wrapper
+ def handleDownloadProgressInfo(self, request: IRequest, content: dict) -> dict:
+ """See ES2+ HandleDownloadProgressInfo in SGP.22 Section 5.3.5"""
+ # spec marks this as FFS (For Further Study)
+ # This is a one way notification from SM-DP+ to Operator, no response
+
+ iccid = content['iccid']
+ profileType = content['profileType']
+ timestamp = content['timestamp']
+ notificationPointId = content['notificationPointId']
+ notificationPointStatus = content['notificationPointStatus']
+
+ # Optional parameters
+ eid = content.get('eid')
+ resultData = content.get('resultData')
+
+ logger.info(f"Download Progress Notification - ICCID: {iccid}, EID: {eid}, "
+ f"ProfileType: {profileType}, Timestamp: {timestamp}, "
+ f"Point: {notificationPointId}, Status: {notificationPointStatus}")
+
+ # OP should process resultData.. probably
+ if resultData:
+ logger.debug(f"Result data received: {resultData}")
+
+ # probably something involving es12 here
+
+ # Per SGP.22, this returns standard success response (empty body)
+ return {}
+
def main(argv):
parser = argparse.ArgumentParser()
@@ -1551,7 +1879,7 @@
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.WARNING)
common_cert_path = os.path.join(DATA_DIR, args.certdir)
- hs = SmDppHttpServer(server_hostname=HOSTNAME, ci_certs_path=os.path.join(common_cert_path, 'CertificateIssuer'), common_cert_path=common_cert_path, use_brainpool=args.brainpool, test_mode=args.test)
+ hs = SmDppHttpServer(server_hostname=HOSTNAME, ci_certs_path=os.path.join(common_cert_path, 'CertificateIssuer'), common_cert_path=common_cert_path, use_brainpool=args.brainpool, in_memory=args.in_memory, test_mode=args.test)
if(args.nossl):
hs.app.run(args.host, args.port)
else:
@@ -1580,10 +1908,84 @@
with open(cert_pempath, 'wb') as pem_file:
pem_file.write(pem_cert)
- SERVER_STRING = f'ssl:{args.port}:privateKey={cert_skpath}:certKey={cert_pempath}:dhParameters={dhparam_path}'
- print(SERVER_STRING)
- hs.app.run(host=HOSTNAME, port=args.port, endpoint_description=SERVER_STRING)
+ # Create custom SSL context factory with client certificate verification
+ class ClientCertContextFactory(ssl.DefaultOpenSSLContextFactory):
+ """SSL Context Factory for ES2+ with SKI-based client certificate trust."""
+
+ def __init__(self, key_path, cert_path, sm_dp_server):
+ super().__init__(key_path, cert_path)
+ self.sm_dp_server = sm_dp_server
+
+ def getContext(self):
+ ctx = SSL.Context(SSL.TLS_METHOD) # Supports TLS 1.2+
+
+ # Load server certificate and private key
+ ctx.use_certificate_file(self.certificateFileName)
+ ctx.use_privatekey_file(self.privateKeyFileName)
+
+ # For ES2+, we request client certificates but don't require them at SSL level
+ # The actual requirement is enforced in the es2plus_api_wrapper
+ flags = SSL.VERIFY_PEER # Request client certificate
+
+ ctx.set_verify(flags, self._verify_callback)
+ ctx.set_verify_depth(10)
+
+ # moah segguriddy
+ ctx.set_options(
+ SSL.OP_NO_SSLv2 |
+ SSL.OP_NO_SSLv3 |
+ SSL.OP_NO_COMPRESSION
+ )
+
+ return ctx
+
+ def _verify_callback(self, connection, x509, errnum, errdepth, ok):
+ """
+ Verify callback for ES2+ client certificates.
+ We don't validate against CAs - instead we check SKI-based trust.
+ """
+ subject = x509.get_subject()
+ cn = getattr(subject, 'commonName', None)
+ org = getattr(subject, 'organizationName', None)
+
+ if errdepth == 0: # Client certificate (not intermediate/root)
+ logger.info(f"[ES2+ VERIFY] Client cert: CN={cn}, O={org}, ok={ok}, errnum={errnum}")
+
+ ski = None
+ try:
+ ski_ext = x509.to_cryptography().extensions.get_extension_for_oid(ExtensionOID.SUBJECT_KEY_IDENTIFIER)
+ ski_bytes = ski_ext.value.key_identifier
+ ski = ':'.join(f'{b:02X}' for b in ski_bytes)
+ logger.info(f"[ES2+ VERIFY] Client SKI: {ski}")
+ except Exception as e:
+ logger.debug(f"[ES2+ VERIFY] Could not extract SKI: {e}")
+
+ # For ES2+ accept all certificates at SSL level, the validation happens in es2plus_api_wrapper so we can respond with proper API error responses
+ if not ok:
+ logger.debug(f"[ES2+ VERIFY] Certificate verification failed (errnum={errnum}), but accepting for API-level handling")
+
+ return True # Always accept at SSL level! I mean it.
+
+ # For intermediate/root certs, accept based on OpenSSL validation
+ return ok
+
+ # Create the SSL context factory following klssl.py pattern
+ contextFactory = ClientCertContextFactory(
+ str(cert_skpath), # key path as string
+ str(cert_pempath), # cert path as string
+ hs # sm_dp_server instance
+ )
+
+ endpoint = SSL4ServerEndpoint(reactor, int(args.port), contextFactory, interface=HOSTNAME)
+
+ site = Site(hs.app.resource())
+ endpoint.listen(site)
+
+ print(f"SM-DP+ server listening on https://{HOSTNAME}:{args.port}")
+ print("ES2+ endpoints require client certificates (SKI-based trust)")
+
+ reactor.run()
if __name__ == "__main__":
main(sys.argv)
diff --git a/pySim/esim/es2plus_commons.py b/pySim/esim/es2plus_commons.py
new file mode 100644
index 0000000..ca73fe7
--- /dev/null
+++ b/pySim/esim/es2plus_commons.py
@@ -0,0 +1,237 @@
+#!/usr/bin/env python3
+# Common validation and utility functions for ES2+ API endpoints.
+#
+# (C) 2025 by Eric Wild <ewild(a)sysmocom.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import shelve
+import logging
+import time
+from typing import Optional, Tuple
+from .smdpp_common import ApiError
+
+logger = logging.getLogger(__name__)
+
+class Es2PlusProfileState:
+ """Encapsulates the state of an ES2+ profile. Tracks profile lifecycle from available through installed."""
+ def __init__(self, iccid: str, profile_type: str = 'Generic', owner: str = 'Unknown'):
+ self.iccid = iccid
+ self.profile_type = profile_type
+ self.owner = owner
+ self.state = 'available' # available, allocated, linked, confirmed, released, downloaded, installed, unavailable
+ self.eid: Optional[str] = None
+ self.matching_id: Optional[str] = None
+ self.confirmation_code_hash: Optional[bytes] = None
+ self.sm_ds_address: Optional[str] = None
+ self.download_attempts: int = 0
+ self.created_timestamp = time.time()
+ self.last_modified = time.time()
+
+ def __getstate__(self):
+ """Helper for pickling to persistent storage."""
+ state = self.__dict__.copy()
+ # All current attributes are pickle-able, but let's be prepared
+ return state
+
+ def __setstate__(self, state):
+ """Helper for unpickling from persistent storage."""
+ self.__dict__.update(state)
+
+class Es2PlusProfileStore:
+ """Database-backed storage for ES2+ profile states using shelve, similar to RspSessionStore."""
+
+ def __init__(self, filename: Optional[str] = None, in_memory: bool = False):
+ self._in_memory = in_memory
+
+ if in_memory:
+ self._shelf = shelve.Shelf(dict())
+ else:
+ if filename is None:
+ raise ValueError("filename is required for file-based profile store")
+ self._shelf = shelve.open(filename)
+
+ # Dictionary-like interface
+ def __getitem__(self, key):
+ return self._shelf[key]
+
+ def __setitem__(self, key, value):
+ value.last_modified = time.time()
+ self._shelf[key] = value
+
+ def __delitem__(self, key):
+ del self._shelf[key]
+
+ def __contains__(self, key):
+ return key in self._shelf
+
+ def __iter__(self):
+ return iter(self._shelf)
+
+ def __len__(self):
+ return len(self._shelf)
+
+ # everything else
+ def __getattr__(self, name):
+ """Delegate attribute access to the underlying shelf object."""
+ return getattr(self._shelf, name)
+
+ def close(self):
+ """Close the session store."""
+ if hasattr(self._shelf, 'close'):
+ self._shelf.close()
+ if self._in_memory:
+ # For in-memory store, clear the reference
+ self._shelf = None
+
+ def sync(self):
+ """Synchronize the cache with the underlying storage."""
+ if hasattr(self._shelf, 'sync'):
+ self._shelf.sync()
+
+ def find_by_matching_id(self, matching_id: str) -> Optional[Es2PlusProfileState]:
+ """Find a profile by its matching ID."""
+ for iccid, profile in self.items():
+ if profile.matching_id == matching_id:
+ return profile
+ return None
+
+ def find_available_by_type(self, profile_type: str, owner: str) -> Optional[Es2PlusProfileState]:
+ """Find first available profile of given type for given owner."""
+ for iccid, profile in self.items():
+ if (profile.state == 'available' and
+ profile.profile_type == profile_type and
+ profile.owner == owner):
+ return profile
+ return None
+
+class Es2PlusHelpers:
+ """Common validation methods for ES2+ operations."""
+
+ @staticmethod
+ def normalize_and_validate_iccid(iccid: str, profile_store) -> Tuple[str, Es2PlusProfileState]:
+ """
+ Normalize ICCID and retrieve profile with validation.
+
+ Returns:
+ Tuple of (normalized_iccid, profile_object)
+ Raises:
+ ApiError if profile not found
+ """
+ # Normalize ICCID (remove F padding)
+ iccid = iccid.rstrip('F')
+
+ # Retrieve profile
+ profile = profile_store.get(iccid)
+ if not profile:
+ # SGP.22 Table: Profile ICCID - Unknown
+ raise ApiError('8.2.1', '3.9', 'Profile unknown', iccid)
+
+ return iccid, profile
+
+ @staticmethod
+ def check_authorization(request, profile, iccid: str, test_mode: bool,
+ profile_store) -> str:
+ """
+ Check authorization and handle ownership assignment.
+
+ Returns:
+ authenticated_entity
+ Raises:
+ ApiError if not authorized
+ """
+ # Get authenticated entity from client certificate
+ authenticated_entity = getattr(request, 'authenticated_entity', None)
+ if not authenticated_entity:
+ # Should not happen if es2plus_api_wrapper is working correctly but this is python, anything is NaN NaN Nan Batman.
+ raise ApiError('8.2.1', '1.2', 'Not authorized - no authenticated entity', iccid)
+
+ # For ES2+, profile ownership is based on the authenticated operator
+ # In test mode, we're kinda flexible with ownership
+ if test_mode:
+ if profile.owner == 'S_MNO' or not profile.owner:
+ profile.owner = authenticated_entity
+ profile_store[iccid] = profile
+ logger.info(f"Test mode: assigned profile {iccid} to {authenticated_entity}")
+ else:
+ # Less lenient in production, strictly check ownership
+ if profile.owner != authenticated_entity:
+ # SGP.22 Table: Profile ICCID - Not Allowed (Authorization)
+ raise ApiError('8.2.1', '1.2', f'Not authorized - profile owned by {profile.owner}', iccid)
+
+ return authenticated_entity
+
+ @staticmethod
+ def validate_eid_association(profile, eid: Optional[str], iccid: str):
+ """
+ Validate EID association with profile.
+
+ Raises:
+ ApiError if EID validation fails
+ """
+ if profile.eid:
+ if not eid:
+ # EID should be provided if associated
+ raise ApiError('8.1.1', '2.2', 'EID required for this order')
+ if eid != profile.eid:
+ # SGP.22 Table: Profile ICCID - Invalid Association
+ raise ApiError('8.2.1', '3.10', 'Different EID associated', iccid)
+
+ @staticmethod
+ def validate_matching_id_association(profile, matching_id: Optional[str]):
+ """
+ Validate matching ID association with profile.
+
+ Raises:
+ ApiError if matching ID validation fails
+ """
+ if profile.matching_id:
+ if matching_id and matching_id != profile.matching_id:
+ # SGP.22 Table: Matching ID - Invalid Association
+ raise ApiError('8.2.6', '3.10', 'Different matchingID associated')
+
+ @staticmethod
+ def handle_sm_ds_event_deletion(profile, sm_ds_events):
+ """Handle SM-DS Event Deletion if needed."""
+ if profile.sm_ds_address and profile.matching_id:
+ # would call ES12.DeleteEvent
+ if profile.matching_id in sm_ds_events:
+ del sm_ds_events[profile.matching_id]
+ logger.info(f"Simulated SM-DS Event Deletion for matchingId: {profile.matching_id}")
+
+ @staticmethod
+ def handle_sm_ds_event_registration(profile, sm_ds_events, server_hostname):
+ """Handle SM-DS Event Registration if needed."""
+ import time
+
+ if (profile.sm_ds_address and
+ profile.matching_id and
+ profile.matching_id not in sm_ds_events):
+ # Simulate ES12.RegisterEvent for now
+ sm_ds_events[profile.matching_id] = {
+ 'eid': profile.eid,
+ 'smdpAddress': server_hostname,
+ 'timestamp': time.time()
+ }
+ logger.info(f"Simulated SM-DS Event Registration for matchingId: {profile.matching_id}")
+
+ @staticmethod
+ def reset_profile_to_available(profile):
+ """Reset profile to available state, clearing all associations."""
+ profile.state = 'available'
+ profile.eid = None
+ profile.matching_id = None
+ profile.confirmation_code_hash = None
+ profile.sm_ds_address = None
+ profile.download_attempts = 0
\ No newline at end of file
diff --git a/pySim/esim/smdpp_common.py b/pySim/esim/smdpp_common.py
new file mode 100644
index 0000000..e1d035e
--- /dev/null
+++ b/pySim/esim/smdpp_common.py
@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+# Common validation and utility functions for the smdpp.
+#
+# (C) 2025 by Eric Wild <ewild(a)sysmocom.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import json
+from typing import Dict, List, Optional
+import logging
+import asn1tools
+from cryptography import x509
+
+logger = logging.getLogger(__name__)
+
+class ApiError(Exception):
+ def __init__(self, subject_code: str, reason_code: str, message: Optional[str] = None,
+ subject_id: Optional[str] = None):
+ self.status_code = build_status_code(subject_code, reason_code, subject_id, message)
+
+ def encode(self) -> str:
+ """Encode the API Error into a responseHeader string."""
+ js = {}
+ build_resp_header(js, 'Failed', self.status_code)
+ return json.dumps(js)
+
+def get_eum_certificate_variant(eum_cert) -> str:
+ """Determine EUM certificate variant by checking Certificate Policies extension.
+ Returns 'O' for old variant, or 'NEW' for Ov3/A/B/C variants."""
+
+ try:
+ cert_policies_ext = eum_cert.extensions.get_extension_for_oid(
+ x509.oid.ExtensionOID.CERTIFICATE_POLICIES
+ )
+
+ for policy in cert_policies_ext.value:
+ policy_oid = policy.policy_identifier.dotted_string
+ logger.debug(f"Found certificate policy: {policy_oid}")
+
+ if policy_oid == '2.23.146.1.2.1.2':
+ logger.debug("Detected EUM certificate variant: O (old)")
+ return 'O'
+ elif policy_oid == '2.23.146.1.2.1.0.0.0':
+ logger.debug("Detected EUM certificate variant: Ov3/A/B/C (new)")
+ return 'NEW'
+ except x509.ExtensionNotFound:
+ logger.debug("No Certificate Policies extension found")
+ except Exception as e:
+ logger.debug(f"Error checking certificate policies: {e}")
+
+def parse_permitted_eins_from_cert(eum_cert) -> List[str]:
+ """Extract permitted IINs from EUM certificate using the appropriate method
+ based on certificate variant (O vs Ov3/A/B/C).
+ Returns list of permitted IINs (basically prefixes that valid EIDs must start with)."""
+
+ # Determine certificate variant first
+ cert_variant = get_eum_certificate_variant(eum_cert)
+ permitted_iins = []
+
+ if cert_variant == 'O':
+ # Old variant - use nameConstraints extension
+ permitted_iins.extend(_parse_name_constraints_eins(eum_cert))
+
+ else:
+ # New variants (Ov3, A, B, C) - use GSMA permittedEins extension
+ permitted_iins.extend(_parse_gsma_permitted_eins(eum_cert))
+
+ unique_iins = list(set(permitted_iins))
+
+ logger.debug(f"Total unique permitted IINs found: {len(unique_iins)}")
+ return unique_iins
+
+def _parse_gsma_permitted_eins(eum_cert) -> List[str]:
+ """Parse the GSMA permittedEins extension using correct ASN.1 structure.
+ PermittedEins ::= SEQUENCE OF PrintableString
+ Each string contains an IIN (Issuer Identification Number) - a prefix of valid EIDs."""
+ permitted_iins = []
+
+ try:
+ permitted_eins_oid = x509.ObjectIdentifier('2.23.146.1.2.2.0') # sgp26: 2.23.146.1.2.2.0 = ASN1:SEQUENCE:permittedEins
+
+ for ext in eum_cert.extensions:
+ if ext.oid == permitted_eins_oid:
+ logger.debug(f"Found GSMA permittedEins extension: {ext.oid}")
+
+ # Get the DER-encoded extension value
+ ext_der = ext.value.value if hasattr(ext.value, 'value') else ext.value
+
+ if isinstance(ext_der, bytes):
+ try:
+ permitted_eins_schema = """
+ PermittedEins DEFINITIONS ::= BEGIN
+ PermittedEins ::= SEQUENCE OF PrintableString
+ END
+ """
+ decoder = asn1tools.compile_string(permitted_eins_schema)
+ decoded_strings = decoder.decode('PermittedEins', ext_der)
+
+ for iin_string in decoded_strings:
+ # Each string contains an IIN -> prefix of euicc EID
+ iin_clean = iin_string.strip().upper()
+
+ # IINs is 8 chars per sgp22, var len according to sgp29, fortunately we don't care
+ if (len(iin_clean) == 8 and
+ all(c in '0123456789ABCDEF' for c in iin_clean) and
+ len(iin_clean) % 2 == 0):
+ permitted_iins.append(iin_clean)
+ logger.debug(f"Found permitted IIN (GSMA): {iin_clean}")
+ else:
+ logger.debug(f"Invalid IIN format: {iin_string} (cleaned: {iin_clean})")
+ except Exception as e:
+ logger.debug(f"Error parsing GSMA permittedEins extension: {e}")
+
+ except Exception as e:
+ logger.debug(f"Error accessing GSMA certificate extensions: {e}")
+
+ return permitted_iins
+
+
+def _parse_name_constraints_eins(eum_cert) -> List[str]:
+ """Parse permitted IINs from nameConstraints extension (variant O)."""
+ permitted_iins = []
+
+ try:
+ # Look for nameConstraints extension
+ name_constraints_ext = eum_cert.extensions.get_extension_for_oid(
+ x509.oid.ExtensionOID.NAME_CONSTRAINTS
+ )
+
+ name_constraints = name_constraints_ext.value
+
+ # Check permittedSubtrees for IIN constraints
+ if name_constraints.permitted_subtrees:
+ for subtree in name_constraints.permitted_subtrees:
+
+ if isinstance(subtree, x509.DirectoryName):
+ for attribute in subtree.value:
+ # IINs for O in serialNumber
+ if attribute.oid == x509.oid.NameOID.SERIAL_NUMBER:
+ serial_value = attribute.value.upper()
+ # sgp22 8, sgp29 var len, fortunately we don't care
+ if (len(serial_value) == 8 and
+ all(c in '0123456789ABCDEF' for c in serial_value) and
+ len(serial_value) % 2 == 0):
+ permitted_iins.append(serial_value)
+ logger.debug(f"Found permitted IIN (nameConstraints/DN): {serial_value}")
+
+ except x509.ExtensionNotFound:
+ logger.debug("No nameConstraints extension found")
+ except Exception as e:
+ logger.debug(f"Error parsing nameConstraints: {e}")
+
+ return permitted_iins
+
+
+def validate_eid_range(eid: str, eum_cert) -> bool:
+ """Validate that EID is within the permitted EINs of the EUM certificate."""
+ if not eid or len(eid) != 32:
+ logger.debug(f"Invalid EID format: {eid}")
+ return False
+
+ try:
+ permitted_eins = parse_permitted_eins_from_cert(eum_cert)
+
+ if not permitted_eins:
+ logger.debug("Warning: No permitted EINs found in EUM certificate")
+ return False
+
+ eid_normalized = eid.upper()
+ logger.debug(f"Validating EID {eid_normalized} against {len(permitted_eins)} permitted EINs")
+
+ for permitted_ein in permitted_eins:
+ if eid_normalized.startswith(permitted_ein):
+ logger.debug(f"EID {eid_normalized} matches permitted EIN {permitted_ein}")
+ return True
+
+ logger.debug(f"EID {eid_normalized} is not in any permitted EIN list")
+ return False
+
+ except Exception as e:
+ logger.debug(f"Error validating EID: {e}")
+ return False
+
+def build_status_code(subject_code: str, reason_code: str, subject_id: Optional[str], message: Optional[str]) -> Dict:
+ r = {'subjectCode': subject_code, 'reasonCode': reason_code }
+ if subject_id:
+ r['subjectIdentifier'] = subject_id
+ if message:
+ r['message'] = message
+ return r
+
+def build_resp_header(js: dict, status: str = 'Executed-Success', status_code_data = None) -> None:
+ # SGP.22 v3.0 6.5.1.4
+ js['header'] = {
+ 'functionExecutionStatus': {
+ 'status': status,
+ }
+ }
+ if status_code_data:
+ js['header']['functionExecutionStatus']['statusCodeData'] = status_code_data
diff --git a/smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned.pem b/smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned.pem
new file mode 100644
index 0000000..02f69e0
--- /dev/null
+++ b/smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned.pem
@@ -0,0 +1,15 @@
+-----BEGIN CERTIFICATE-----
+MIICQTCCAeagAwIBAgIUBH9i6K/qKONp3qK84jbPOeYJ3q8wCgYIKoZIzj0EAwIw
+YzEiMCAGA1UEAwwZVEVTVF9PUEVSQVRPUiBFUzIrIENsaWVudDEWMBQGA1UECgwN
+VEVTVF9PUEVSQVRPUjEYMBYGA1UECwwPZVNJTSBPcGVyYXRpb25zMQswCQYDVQQG
+EwJVUzAeFw0yNTA3MjcyMDAyNTBaFw0yNjA3MjgyMDAyNTBaMGMxIjAgBgNVBAMM
+GVRFU1RfT1BFUkFUT1IgRVMyKyBDbGllbnQxFjAUBgNVBAoMDVRFU1RfT1BFUkFU
+T1IxGDAWBgNVBAsMD2VTSU0gT3BlcmF0aW9uczELMAkGA1UEBhMCVVMwWTATBgcq
+hkjOPQIBBggqhkjOPQMBBwNCAAQsjlrne8IklFEW3BGu3FEJ0EEdwkE/vf2oqr+T
+l8EFqkbKIDsu0yH+LmGYxXAvQjEXa8GvTwG9aln/VUcdt/oto3gwdjAdBgNVHQ4E
+FgQUGZRJ3dJ/Ks7Q6BIjLBoKcgueJWowHwYDVR0jBBgwFoAUGZRJ3dJ/Ks7Q6BIj
+LBoKcgueJWowDAYDVR0TAQH/BAIwADAOBgNVHQ8BAf8EBAMCBaAwFgYDVR0lAQH/
+BAwwCgYIKwYBBQUHAwIwCgYIKoZIzj0EAwIDSQAwRgIhAMFb61Bya2aiF7X7TTXm
+alo0rffrzPdeDL0I4HeHEddQAiEA1kfH4AdlBYl0u+tgLqqHMcjiI6OrT6ywUsYD
+p6q408U=
+-----END CERTIFICATE-----
diff --git a/smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned_combined.pem b/smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned_combined.pem
new file mode 100644
index 0000000..2ffec8b
--- /dev/null
+++ b/smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned_combined.pem
@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIICQTCCAeagAwIBAgIUBH9i6K/qKONp3qK84jbPOeYJ3q8wCgYIKoZIzj0EAwIw
+YzEiMCAGA1UEAwwZVEVTVF9PUEVSQVRPUiBFUzIrIENsaWVudDEWMBQGA1UECgwN
+VEVTVF9PUEVSQVRPUjEYMBYGA1UECwwPZVNJTSBPcGVyYXRpb25zMQswCQYDVQQG
+EwJVUzAeFw0yNTA3MjcyMDAyNTBaFw0yNjA3MjgyMDAyNTBaMGMxIjAgBgNVBAMM
+GVRFU1RfT1BFUkFUT1IgRVMyKyBDbGllbnQxFjAUBgNVBAoMDVRFU1RfT1BFUkFU
+T1IxGDAWBgNVBAsMD2VTSU0gT3BlcmF0aW9uczELMAkGA1UEBhMCVVMwWTATBgcq
+hkjOPQIBBggqhkjOPQMBBwNCAAQsjlrne8IklFEW3BGu3FEJ0EEdwkE/vf2oqr+T
+l8EFqkbKIDsu0yH+LmGYxXAvQjEXa8GvTwG9aln/VUcdt/oto3gwdjAdBgNVHQ4E
+FgQUGZRJ3dJ/Ks7Q6BIjLBoKcgueJWowHwYDVR0jBBgwFoAUGZRJ3dJ/Ks7Q6BIj
+LBoKcgueJWowDAYDVR0TAQH/BAIwADAOBgNVHQ8BAf8EBAMCBaAwFgYDVR0lAQH/
+BAwwCgYIKwYBBQUHAwIwCgYIKoZIzj0EAwIDSQAwRgIhAMFb61Bya2aiF7X7TTXm
+alo0rffrzPdeDL0I4HeHEddQAiEA1kfH4AdlBYl0u+tgLqqHMcjiI6OrT6ywUsYD
+p6q408U=
+-----END CERTIFICATE-----
+-----BEGIN EC PRIVATE KEY-----
+MHcCAQEEIA8rDic3h4Vc4aM8U81daYRxr5UphTWpEn3mWQPeViTSoAoGCCqGSM49
+AwEHoUQDQgAELI5a53vCJJRRFtwRrtxRCdBBHcJBP739qKq/k5fBBapGyiA7LtMh
+/i5hmMVwL0IxF2vBr08BvWpZ/1VHHbf6LQ==
+-----END EC PRIVATE KEY-----
diff --git a/smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned_key.pem b/smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned_key.pem
new file mode 100644
index 0000000..687a609
--- /dev/null
+++ b/smdpp-data/certs/SelfSignedOperators/TEST_OPERATOR_selfsigned_key.pem
@@ -0,0 +1,5 @@
+-----BEGIN EC PRIVATE KEY-----
+MHcCAQEEIA8rDic3h4Vc4aM8U81daYRxr5UphTWpEn3mWQPeViTSoAoGCCqGSM49
+AwEHoUQDQgAELI5a53vCJJRRFtwRrtxRCdBBHcJBP739qKq/k5fBBapGyiA7LtMh
+/i5hmMVwL0IxF2vBr08BvWpZ/1VHHbf6LQ==
+-----END EC PRIVATE KEY-----
diff --git a/tests/unittests/test_es2plus.py b/tests/unittests/test_es2plus.py
new file mode 100644
index 0000000..da38c02
--- /dev/null
+++ b/tests/unittests/test_es2plus.py
@@ -0,0 +1,244 @@
+#!/usr/bin/env python3
+# Integrated test for ES2+ functionality using module imports of our existing cli scripts (es2p client, cert gen).
+#
+# (C) 2025 by Eric Wild <ewild(a)sysmocom.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import socket
+import sys
+import tempfile
+import unittest
+import subprocess
+import time
+import shutil
+from pathlib import Path
+
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+sys.path.insert(0, str(project_root / 'contrib'))
+
+from generate_self_signed_operator_cert import generate_self_signed_operator_cert # noqa: E402
+from pySim.esim.es2p import Es2pApiClient # noqa: E402
+
+class HostnameResolutionMixin:
+ """Mixin to fix cert addresses to localhost"""
+
+ def setUp(self):
+ super().setUp()
+
+ self.hostname_mappings = getattr(self, 'hostname_mappings', {
+ 'testsmdpplus1.example.com': 'localhost'
+ })
+
+ self._original_getaddrinfo = socket.getaddrinfo
+ def patched_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
+ if host in self.hostname_mappings:
+ target_host = self.hostname_mappings[host]
+ return self._original_getaddrinfo(target_host, port, family, type, proto, flags)
+ return self._original_getaddrinfo(host, port, family, type, proto, flags)
+ socket.getaddrinfo = patched_getaddrinfo
+
+ def tearDown(self):
+ socket.getaddrinfo = self._original_getaddrinfo
+ super().tearDown()
+
+class TestES2PlusDemo(HostnameResolutionMixin, unittest.TestCase):
+ """Test ES2+ mutual TLS authentication with self-signed certificates."""
+
+ @classmethod
+ def setUpClass(cls):
+ """Set up test environment once for all tests."""
+ cls.project_root = Path(__file__).parent.parent.parent
+ cls.test_dir = tempfile.mkdtemp(prefix='es2p_test_')
+ cls.cert_dir = os.path.join(cls.test_dir, 'certs')
+ os.makedirs(cls.cert_dir, exist_ok=True)
+
+ cls.server_ca_cert = str(cls.project_root / 'smdpp-data' / 'certs' /
+ 'CertificateIssuer' / 'CERT_CI_ECDSA_NIST.pem')
+
+ @classmethod
+ def tearDownClass(cls):
+ """Clean up test environment."""
+ if hasattr(cls, 'test_dir') and os.path.exists(cls.test_dir):
+ shutil.rmtree(cls.test_dir)
+
+ def setUp(self):
+ """Set up each test."""
+ super().setUp()
+ self.server_process = None
+
+ def tearDown(self):
+ """Clean up after each test."""
+ if self.server_process:
+ self.server_process.terminate()
+ try:
+ self.server_process.wait(timeout=2)
+ except subprocess.TimeoutExpired:
+ self.server_process.kill()
+ self.server_process.wait()
+
+ if self.server_process.stdout:
+ self.server_process.stdout.close()
+ if self.server_process.stderr:
+ self.server_process.stderr.close()
+ super().tearDown()
+
+
+ def start_smdpp_server(self):
+ """Start the SM-DP+ server as a subprocess (like in shell script)."""
+ cmd = [
+ 'python3',
+ str(self.project_root / 'osmo-smdpp.py'),
+ '-t', # test mode
+ '-m', # in-memory storage
+ ]
+
+ self.server_process = subprocess.Popen(
+ cmd,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ cwd=str(self.project_root)
+ )
+
+ # Wait for server to start because.. python
+ time.sleep(3)
+
+ def test_self_signed_certificate_flow(self):
+ """Test ES2+ downloadOrder with self-signed operator certificate.
+ Steps:
+ 1. Generation of self-signed operator certificates
+ 2. Direct use of Es2pApiClient from imported cli module
+ 3. Mutual TLS authentication with self-signed certificates
+ 4. Trust establishment through SKI at app layer
+ """
+ # Step 1: Generate self-signed operator certificate using imported function
+ operator_name = 'TEST_OPERATOR'
+ cert_path, ski = generate_self_signed_operator_cert(operator_name, self.cert_dir)
+
+ # Verify certificate was created
+ self.assertTrue(os.path.exists(cert_path))
+ self.assertIsNotNone(ski)
+ self.assertIn(':', ski) # SKI should be formatted with colons
+
+ # Step 2: Start SM-DP+ server
+ self.start_smdpp_server()
+
+ try:
+ client = Es2pApiClient(
+ url_prefix='https://testsmdpplus1.example.com:8000/gsma/rsp2/es2plus',
+ func_req_id=operator_name,
+ server_cert_verify=self.server_ca_cert,
+ client_cert=cert_path
+ )
+
+ # Step 4: Make downloadOrder request
+ request_data = {
+ 'profileType': 'Test'
+ }
+
+ result = client.call_downloadOrder(request_data)
+
+ self.assertIsInstance(result, dict)
+
+ # In test mode, the server should accept the self-signed certificate and return a response
+ # - even if it's an error due to no matching profiles, the whole point is that the TLS handshake succeeds
+
+ # Check if we got a functionExecutionStatus (expected in response)
+ if 'functionExecutionStatus' in result:
+ status = result['functionExecutionStatus']
+ self.assertIn('status', status)
+ # May be 'Executed-Success' or 'Failed' depending on profile availability, don't care
+ self.assertIn(status['status'], ['Executed-Success', 'Failed'])
+
+ except Exception as e:
+ # If we get an SSL error, the test failed
+ if 'SSL' in str(e) or 'certificate' in str(e).lower():
+ self.fail(f"SSL/Certificate error occurred: {e}")
+
+ def test_multiple_operator_certificates(self):
+ """Test that different operators can use different self-signed certificates."""
+ operators = ['OPERATOR_A', 'OPERATOR_B', 'OPERATOR_C']
+ certificates = {}
+
+ # Generate certificates for each operator
+ for operator in operators:
+ cert_path, ski = generate_self_signed_operator_cert(operator, self.cert_dir)
+ certificates[operator] = {
+ 'cert_path': cert_path,
+ 'ski': ski
+ }
+ self.assertTrue(os.path.exists(cert_path))
+
+ self.start_smdpp_server()
+
+ # Test that each operator can authenticate with their own certificate
+ for operator, cert_info in certificates.items():
+ client = Es2pApiClient(
+ url_prefix='https://testsmdpplus1.example.com:8000/gsma/rsp2/es2plus',
+ func_req_id=operator,
+ server_cert_verify=self.server_ca_cert,
+ client_cert=cert_info['cert_path']
+ )
+
+ request_data = {'profileType': f'Test_{operator}'}
+
+ try:
+ result = client.call_downloadOrder(request_data)
+ self.assertIsInstance(result, dict)
+ # As above only TLS handshake matters
+ except Exception as e:
+ if 'SSL' in str(e) or 'certificate' in str(e).lower():
+ self.fail(f"SSL/Certificate error for {operator}: {e}")
+
+ def test_client_without_certificate_fails(self):
+ """Test that clients without certificates are rejected when mutual TLS is required."""
+ self.start_smdpp_server()
+
+ client = Es2pApiClient(
+ url_prefix='https://testsmdpplus1.example.com:8000/gsma/rsp2/es2plus',
+ func_req_id='NO_CERT_CLIENT',
+ server_cert_verify=self.server_ca_cert,
+ client_cert=None # No client certificate
+ )
+
+ request_data = {'profileType': 'Test'}
+
+ # This should fail due to missing client certificate
+ # In test mode, the server may not enforce client certificates at TLS layer but rejects at application layer
+ try:
+ result = client.call_downloadOrder(request_data)
+ # If we get here, check if server rejected at application layer
+ if 'functionExecutionStatus' in result:
+ status = result['functionExecutionStatus']
+ # Should be rejected due to missing authentication
+ self.assertEqual(status.get('status'), 'Failed', "Expected failure due to missing client certificate")
+ else:
+ # Got some response but no clear rejection (?!)
+ self.fail("Server should reject requests without client certificates")
+ except Exception as e:
+ # SSL/TLS layer rejection is also acceptable
+ error_msg = str(e)
+ self.assertTrue(
+ 'certificate' in error_msg.lower() or
+ 'SSL' in error_msg or
+ 'handshake' in error_msg.lower() or
+ '404' in error_msg, # Server might return 404 for unauthenticated requests
+ f"Expected certificate or authentication error, got: {error_msg}"
+ )
+
+
+if __name__ == '__main__':
+ unittest.main(verbosity=2)
\ No newline at end of file
diff --git a/tests/unittests/test_es2plus_es9plus_http.py b/tests/unittests/test_es2plus_es9plus_http.py
new file mode 100644
index 0000000..8324ab9
--- /dev/null
+++ b/tests/unittests/test_es2plus_es9plus_http.py
@@ -0,0 +1,582 @@
+#!/usr/bin/env python3
+# Consolidated test suite for ES2+/ES9+ integration.
+# HTTP endpoint testing with actual profile storage testing
+#
+# (C) 2025 by Eric Wild <ewild(a)sysmocom.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import json
+import base64
+import sys
+import os
+import tempfile
+import hashlib
+import unittest
+import importlib.util
+from unittest.mock import patch, MagicMock
+from pathlib import Path
+
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from pySim.utils import h2b # noqa: E402
+
+
+def import_osmo_smdpp():
+ """Import osmo-smdpp module dynamically."""
+ project_root = Path(__file__).parent.parent.parent
+ module_path = project_root / 'osmo-smdpp.py'
+ if 'osmo_smdpp' in sys.modules:
+ return sys.modules['osmo_smdpp']
+
+ spec = importlib.util.spec_from_file_location('osmo_smdpp', module_path)
+ if spec is None or spec.loader is None:
+ raise ImportError(f"Cannot load module from {module_path}")
+
+ module = importlib.util.module_from_spec(spec)
+
+ # Register it in sys.modules with underscores so pickle will look for 'osmo_smdpp' not 'osmo-smdpp'
+ sys.modules['osmo_smdpp'] = module
+ spec.loader.exec_module(module)
+ return module
+
+
+osmo_smdpp = import_osmo_smdpp()
+
+Es2PlusProfileState = osmo_smdpp.Es2PlusProfileState
+Es2PlusProfileStore = osmo_smdpp.Es2PlusProfileStore
+SmDppHttpServer = osmo_smdpp.SmDppHttpServer
+
+
+class ES2PlusHTTPEndpointTest(unittest.TestCase):
+ """Test ES2+ and ES9+ HTTP endpoints with mocked responses."""
+
+ def setUp(self):
+ """Set up test environment."""
+ self.smdp_host = "localhost"
+ self.smdp_port = 8000
+ self.base_url = f"http://{self.smdp_host}:{self.smdp_port}"
+
+ self.test_eid = "89001012012341234012345678901234"
+ self.test_iccid = "8900000000000000001"
+ self.test_matching_id = None # Will be set by mock confirmOrder
+ self.test_confirmation_code = "12345678"
+
+ def _create_mock_response(self, status_code=200, json_data=None):
+ """Create a mock HTTP response."""
+ mock_response = MagicMock()
+ mock_response.status_code = status_code
+ mock_response.raise_for_status.return_value = None
+ if json_data:
+ mock_response.json.return_value = json_data
+ return mock_response
+
+ @patch('requests.post')
+ def test_es2plus_download_order(self, mock_post):
+ """Test ES2+ DownloadOrder HTTP endpoint."""
+ # Mock successful response
+ mock_response_data = {
+ "header": {
+ "functionExecutionStatus": {
+ "status": "Executed-Success"
+ }
+ },
+ "iccid": self.test_iccid
+ }
+ mock_post.return_value = self._create_mock_response(200, mock_response_data)
+
+ # Import requests here since we're mocking it
+ import requests
+
+ data = {
+ "eid": self.test_eid,
+ "iccid": self.test_iccid,
+ "profileType": "Test"
+ }
+
+ url = f"{self.base_url}/gsma/rsp2/es2plus/downloadOrder"
+ headers = {"Content-Type": "application/json"}
+ response = requests.post(url, json=data, headers=headers)
+ result = response.json()
+
+ mock_post.assert_called_once_with(url, json=data, headers=headers)
+
+ self.assertEqual(result["header"]["functionExecutionStatus"]["status"], "Executed-Success")
+ self.assertEqual(result["iccid"], self.test_iccid)
+
+ @patch('requests.post')
+ def test_es2plus_confirm_order(self, mock_post):
+ """Test ES2+ ConfirmOrder HTTP endpoint."""
+ test_matching_id = "TEST_MATCHING_ID_12345"
+
+ # Mock successful response
+ mock_response_data = {
+ "header": {
+ "functionExecutionStatus": {
+ "status": "Executed-Success"
+ }
+ },
+ "matchingId": test_matching_id
+ }
+ mock_post.return_value = self._create_mock_response(200, mock_response_data)
+
+ import requests
+
+ data = {
+ "iccid": self.test_iccid,
+ "releaseFlag": True,
+ "eid": self.test_eid,
+ "confirmationCode": self.test_confirmation_code
+ }
+
+ url = f"{self.base_url}/gsma/rsp2/es2plus/confirmOrder"
+ headers = {"Content-Type": "application/json"}
+ response = requests.post(url, json=data, headers=headers)
+ result = response.json()
+
+ mock_post.assert_called_once_with(url, json=data, headers=headers)
+
+ self.assertEqual(result["header"]["functionExecutionStatus"]["status"], "Executed-Success")
+ self.assertEqual(result["matchingId"], test_matching_id)
+
+ @patch('requests.post')
+ def test_es9plus_initiate_authentication(self, mock_post):
+ """Test ES9+ InitiateAuthentication HTTP endpoint."""
+ test_transaction_id = "TXN_12345"
+
+ # Mock successful response
+ mock_response_data = {
+ "header": {
+ "functionExecutionStatus": {
+ "status": "Executed-Success"
+ }
+ },
+ "transactionId": test_transaction_id
+ }
+ mock_post.return_value = self._create_mock_response(200, mock_response_data)
+
+ import requests
+
+ euicc_info1 = {
+ "svn": "2.2.0",
+ "euiccCiPKIdListForVerification": [
+ {"SubjectKeyIdentifier": "F54172BDF98A95D65CBEB88A38A1C11D800A85C3"}
+ ],
+ "euiccCiPKIdListForSigning": [
+ {"SubjectKeyIdentifier": "F54172BDF98A95D65CBEB88A38A1C11D800A85C3"}
+ ]
+ }
+
+ euicc_info1_b64 = base64.b64encode(json.dumps(euicc_info1).encode()).decode()
+ data = {
+ "euiccChallenge": base64.b64encode(b"test_challenge_123").decode(),
+ "euiccInfo1": euicc_info1_b64,
+ "smdpAddress": f"{self.smdp_host}:{self.smdp_port}"
+ }
+
+ url = f"{self.base_url}/gsma/rsp2/es9plus/initiateAuthentication"
+ headers = {"Content-Type": "application/json"}
+ response = requests.post(url, json=data, headers=headers)
+ result = response.json()
+
+ mock_post.assert_called_once_with(url, json=data, headers=headers)
+
+ self.assertEqual(result["header"]["functionExecutionStatus"]["status"], "Executed-Success")
+ self.assertEqual(result["transactionId"], test_transaction_id)
+
+ @patch('requests.post')
+ def test_es9plus_authenticate_client_with_matching_id(self, mock_post):
+ """Test ES9+ AuthenticateClient with matchingId."""
+ test_transaction_id = "TXN_12345"
+ test_matching_id = "TEST_MATCHING_ID_12345"
+
+ # mock a failure due to certificate validation but ensure that the matchingId was processed
+ mock_response = MagicMock()
+ mock_response.status_code = 400
+ mock_response.raise_for_status.side_effect = Exception("Certificate validation failed")
+ mock_response.text = "matchingId processed but certificate validation failed"
+ mock_post.return_value = mock_response
+
+ import requests
+
+ authenticate_server_response = {
+ "authenticateResponseOk": {
+ "euiccSigned1": {
+ "transactionId": test_transaction_id,
+ "serverChallenge": "mock_server_challenge",
+ "ctxParams1": {
+ "ctxParamsForCommonAuthentication": {
+ "matchingId": test_matching_id
+ }
+ }
+ },
+ "euiccSignature1": base64.b64encode(b"mock_signature").decode(),
+ "euiccCertificate": "mock_certificate",
+ "eumCertificate": "mock_eum_certificate"
+ }
+ }
+
+ data = {
+ "transactionId": test_transaction_id,
+ "authenticateServerResponse": base64.b64encode(
+ json.dumps(authenticate_server_response).encode()
+ ).decode()
+ }
+
+ # Make request and expect it to fail (due to mock certificates)
+ url = f"{self.base_url}/gsma/rsp2/es9plus/authenticateClient"
+ headers = {"Content-Type": "application/json"}
+
+ with self.assertRaises(Exception) as context:
+ requests.post(url, json=data, headers=headers)
+ response = context.exception.response if hasattr(context.exception, 'response') else mock_response
+ response.raise_for_status()
+
+ mock_post.assert_called_once_with(url, json=data, headers=headers)
+
+ # Verify the matchingId was included in the request
+ call_args = mock_post.call_args
+ request_data = call_args[1]['json']
+
+ encoded_response = request_data['authenticateServerResponse']
+ decoded_response = json.loads(base64.b64decode(encoded_response).decode())
+
+ matching_id_in_request = (decoded_response['authenticateResponseOk']
+ ['euiccSigned1']['ctxParams1']
+ ['ctxParamsForCommonAuthentication']['matchingId'])
+
+ self.assertEqual(matching_id_in_request, test_matching_id)
+
+
+class ES2PlusProfileIntegrationTest(unittest.TestCase):
+ """Test ES2+ profile storage and ES9+ integration."""
+
+ def setUp(self):
+ """Set up test environment."""
+ self.test_iccid = "8900000000000000099"
+ self.test_matching_id = "TEST_ES2PLUS_MID_123"
+ self.test_confirmation_code = "87654321"
+ self.test_eid = "89001012012341234012345678901234"
+
+ self.temp_dir = tempfile.mkdtemp()
+ self.profile_store_path = os.path.join(self.temp_dir, "test-es2plus-profiles")
+ self.profile_store = Es2PlusProfileStore(filename=self.profile_store_path)
+
+ def test_es2plus_profile_creation_and_storage(self):
+ """Test creating and storing ES2+ profiles."""
+ # Create an ES2+ profile as if it was provisioned via downloadOrder/confirmOrder
+ profile = Es2PlusProfileState(self.test_iccid, 'Test', 'TestOperator')
+ profile.matching_id = self.test_matching_id
+ profile.state = 'released'
+ profile.eid = self.test_eid
+
+ # Store confirmation code hash (SHA256 of the code)
+ cc_bytes = h2b(self.test_confirmation_code)
+ profile.confirmation_code_hash = hashlib.sha256(cc_bytes).digest()
+
+ self.profile_store[self.test_iccid] = profile
+ self.profile_store.sync()
+
+ self.assertIn(self.test_iccid, self.profile_store)
+ stored_profile = self.profile_store[self.test_iccid]
+ self.assertEqual(stored_profile.state, 'released')
+ self.assertEqual(stored_profile.matching_id, self.test_matching_id)
+ self.assertIsNotNone(stored_profile.confirmation_code_hash)
+
+ def test_profile_lookup_by_matching_id(self):
+ """Test profile lookup by matching ID."""
+ profile = Es2PlusProfileState(self.test_iccid, 'Test', 'TestOperator')
+ profile.matching_id = self.test_matching_id
+ profile.state = 'released'
+ profile.eid = self.test_eid
+
+ cc_bytes = h2b(self.test_confirmation_code)
+ profile.confirmation_code_hash = hashlib.sha256(cc_bytes).digest()
+
+ self.profile_store[self.test_iccid] = profile
+ self.profile_store.sync()
+
+ # Test lookup by matching_id
+ found_profile = self.profile_store.find_by_matching_id(self.test_matching_id)
+
+ self.assertIsNotNone(found_profile, "Profile not found by matching_id")
+ self.assertEqual(found_profile.iccid, self.test_iccid)
+ self.assertEqual(found_profile.state, 'released')
+ self.assertEqual(found_profile.eid, self.test_eid)
+ self.assertIsNotNone(found_profile.confirmation_code_hash)
+
+ def test_server_integration_with_es2plus_profiles(self):
+ """Test server's ability to find ES2+ profiles."""
+ profile = Es2PlusProfileState(self.test_iccid, 'Test', 'TestOperator')
+ profile.matching_id = self.test_matching_id
+ profile.state = 'released'
+ profile.eid = self.test_eid
+
+ cc_bytes = h2b(self.test_confirmation_code)
+ profile.confirmation_code_hash = hashlib.sha256(cc_bytes).digest()
+
+ self.profile_store[self.test_iccid] = profile
+ self.profile_store.sync()
+
+ server = SmDppHttpServer(
+ server_hostname='testsmdpplus1.example.com',
+ ci_certs_path=str(project_root / 'smdpp-data/certs/CertificateIssuer'),
+ common_cert_path=str(project_root / 'smdpp-data/certs'),
+ use_brainpool=False,
+ in_memory=True,
+ test_mode=True
+ )
+
+ # Replace server's profile store with our test store
+ server.profile_store = self.profile_store
+
+ # Test server's ability to find the profile
+ es2plus_profile = server.profile_store.find_by_matching_id(self.test_matching_id)
+
+ self.assertIsNotNone(es2plus_profile, "Server cannot find ES2+ profile")
+ self.assertEqual(es2plus_profile.iccid, self.test_iccid)
+ self.assertEqual(es2plus_profile.state, 'released')
+ self.assertEqual(es2plus_profile.matching_id, self.test_matching_id)
+ self.assertEqual(es2plus_profile.eid, self.test_eid)
+
+ def test_es9plus_would_accept_profile(self):
+ """Test that ES9+ logic would accept the ES2+ profile."""
+ profile = Es2PlusProfileState(self.test_iccid, 'Test', 'TestOperator')
+ profile.matching_id = self.test_matching_id
+ profile.state = 'released'
+ profile.eid = self.test_eid
+
+ cc_bytes = h2b(self.test_confirmation_code)
+ profile.confirmation_code_hash = hashlib.sha256(cc_bytes).digest()
+
+ self.profile_store[self.test_iccid] = profile
+ self.profile_store.sync()
+
+ # Find the profile (simulating ES9+ authenticateClient logic)
+ found_profile = self.profile_store.find_by_matching_id(self.test_matching_id)
+
+ self.assertIsNotNone(found_profile)
+ self.assertEqual(found_profile.state, 'released',
+ "Profile must be in 'released' state for ES9+ download")
+ self.assertIsNotNone(found_profile.confirmation_code_hash,
+ "Profile should have confirmation code hash")
+
+ expected_hash = hashlib.sha256(cc_bytes).digest()
+ self.assertEqual(found_profile.confirmation_code_hash, expected_hash,
+ "Confirmation code hash mismatch")
+
+ def test_activation_code_profiles_sync(self):
+ """Test that ES2+ profiles are synced to activation_code_profiles."""
+ server = SmDppHttpServer(
+ server_hostname='testsmdpplus1.example.com',
+ ci_certs_path=str(project_root / 'smdpp-data/certs/CertificateIssuer'),
+ common_cert_path=str(project_root / 'smdpp-data/certs'),
+ use_brainpool=False,
+ in_memory=True,
+ test_mode=True
+ )
+
+ test_matching_id = "ES2PLUS_SYNC_TEST"
+ test_iccid = "8900000000000000088"
+
+ profile = Es2PlusProfileState(test_iccid, 'Test', 'TestOp')
+ profile.matching_id = test_matching_id
+ profile.state = 'released'
+ server.profile_store[profile.iccid] = profile
+
+ # In test mode, manually trigger sync (simulating confirmOrder behavior)
+ if server.test_mode:
+ server.activation_code_profiles[test_matching_id] = {
+ 'matchingId': test_matching_id,
+ 'confirmationCode': None,
+ 'iccid': profile.iccid,
+ 'profileName': f'ES2+ Profile {profile.iccid}',
+ 'state': 'released',
+ 'download_attempts': 0,
+ 'cc_attempts': 0,
+ 'associated_eid': None,
+ 'expiration': None,
+ 'profile_path': 'TS48v4_SAIP2.3_BERTLV'
+ }
+
+ self.assertIn(test_matching_id, server.activation_code_profiles,
+ "ES2+ profile should be synced to activation_code_profiles")
+
+ synced_profile = server.activation_code_profiles[test_matching_id]
+ self.assertEqual(synced_profile['iccid'], test_iccid)
+ self.assertEqual(synced_profile['state'], 'released')
+
+ def tearDown(self):
+ """Clean up test environment."""
+ if hasattr(self, 'profile_store'):
+ self.profile_store.close()
+
+ if hasattr(self, 'temp_dir'):
+ import shutil
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+
+class ES2PlusProfileStoreTest(unittest.TestCase):
+ """Test ES2PlusProfileStore functionality."""
+
+ def setUp(self):
+ """Set up test environment."""
+ self.temp_dir = tempfile.mkdtemp()
+ self.store_path = os.path.join(self.temp_dir, "test-store")
+ self.store = Es2PlusProfileStore(filename=self.store_path)
+
+ def test_profile_store_basic_operations(self):
+ """Test basic profile store operations."""
+ iccid = "8900000000000000001"
+ profile = Es2PlusProfileState(iccid, "Test", "TestOp")
+ profile.matching_id = "TEST_BASIC"
+ profile.state = "allocated"
+
+ self.store[iccid] = profile
+
+ self.assertIn(iccid, self.store)
+ retrieved = self.store[iccid]
+ self.assertEqual(retrieved.iccid, iccid)
+ self.assertEqual(retrieved.matching_id, "TEST_BASIC")
+ self.assertEqual(retrieved.state, "allocated")
+
+ def test_find_by_matching_id_functionality(self):
+ """Test find_by_matching_id method."""
+ profiles = [
+ ("8900000000000000001", "MATCH_ID_1", "released"),
+ ("8900000000000000002", "MATCH_ID_2", "confirmed"),
+ ("8900000000000000003", "MATCH_ID_3", "allocated"),
+ ]
+
+ for iccid, match_id, state in profiles:
+ profile = Es2PlusProfileState(iccid, "Test", "TestOp")
+ profile.matching_id = match_id
+ profile.state = state
+ self.store[iccid] = profile
+
+ for iccid, match_id, state in profiles:
+ with self.subTest(matching_id=match_id):
+ found = self.store.find_by_matching_id(match_id)
+ self.assertIsNotNone(found)
+ self.assertEqual(found.iccid, iccid)
+ self.assertEqual(found.matching_id, match_id)
+ self.assertEqual(found.state, state)
+
+ not_found = self.store.find_by_matching_id("NON_EXISTENT")
+ self.assertIsNone(not_found)
+
+ def tearDown(self):
+ """Clean up test environment."""
+ self.store.close()
+ import shutil
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+
+class ES2PlusWorkflowTest(unittest.TestCase):
+ """Test complete ES2+ to ES9+ workflow."""
+
+ def setUp(self):
+ """Set up test environment for workflow tests."""
+ self.temp_dir = tempfile.mkdtemp()
+ self.profile_store_path = os.path.join(self.temp_dir, "workflow-es2plus-profiles")
+ self.profile_store = Es2PlusProfileStore(filename=self.profile_store_path)
+
+ self.test_eid = "89001012012341234012345678901234"
+ self.test_iccid = "8900000000000000001"
+ self.test_confirmation_code = "12345678"
+ self.test_matching_id = "WORKFLOW_TEST_MID"
+
+ def test_complete_workflow_with_profile_store(self):
+ """Test complete workflow using actual profile store."""
+ # Step 1: Simulate DownloadOrder - create profile
+ profile = Es2PlusProfileState(self.test_iccid, 'Test', 'TestOperator')
+ profile.state = 'allocated'
+ self.profile_store[self.test_iccid] = profile
+ self.profile_store.sync()
+
+ stored_profile = self.profile_store[self.test_iccid]
+ self.assertEqual(stored_profile.state, 'allocated')
+
+ # Step 2: Simulate ConfirmOrder - update profile with matching_id
+ stored_profile.matching_id = self.test_matching_id
+ stored_profile.state = 'released'
+ stored_profile.eid = self.test_eid
+
+ cc_bytes = h2b(self.test_confirmation_code)
+ stored_profile.confirmation_code_hash = hashlib.sha256(cc_bytes).digest()
+
+ self.profile_store[self.test_iccid] = stored_profile
+ self.profile_store.sync()
+
+ confirmed_profile = self.profile_store[self.test_iccid]
+ self.assertEqual(confirmed_profile.state, 'released')
+ self.assertEqual(confirmed_profile.matching_id, self.test_matching_id)
+
+ # Step 3: Simulate ES9+ lookup by matching_id
+ es9_profile = self.profile_store.find_by_matching_id(self.test_matching_id)
+
+ self.assertIsNotNone(es9_profile)
+ self.assertEqual(es9_profile.iccid, self.test_iccid)
+ self.assertEqual(es9_profile.state, 'released')
+ self.assertEqual(es9_profile.eid, self.test_eid)
+
+ # Step 4: Verify complete data integrity
+ self.assertIsNotNone(es9_profile.confirmation_code_hash)
+ expected_hash = hashlib.sha256(cc_bytes).digest()
+ self.assertEqual(es9_profile.confirmation_code_hash, expected_hash)
+
+ def test_error_handling_workflow(self):
+ """Test error handling in the integration workflow."""
+ # Test 1: Profile not found by matching_id
+ non_existent_profile = self.profile_store.find_by_matching_id("NON_EXISTENT_MID")
+ self.assertIsNone(non_existent_profile)
+
+ # Test 2: Profile in wrong state for ES9+ download
+ profile = Es2PlusProfileState(self.test_iccid, 'Test', 'TestOperator')
+ profile.matching_id = "ERROR_TEST_MID"
+ profile.state = 'allocated' # Wrong state - should be 'released'
+ self.profile_store[self.test_iccid] = profile
+ self.profile_store.sync()
+
+ error_profile = self.profile_store.find_by_matching_id("ERROR_TEST_MID")
+ self.assertIsNotNone(error_profile)
+ self.assertNotEqual(error_profile.state, 'released',
+ "Profile should not be in released state for error test")
+
+ # Test 3: Missing confirmation code hash
+ profile2 = Es2PlusProfileState("8900000000000000002", 'Test', 'TestOperator')
+ profile2.matching_id = "NO_CC_HASH_MID"
+ profile2.state = 'released'
+ # Intentionally not setting confirmation_code_hash
+ self.profile_store[profile2.iccid] = profile2
+ self.profile_store.sync()
+
+ no_cc_profile = self.profile_store.find_by_matching_id("NO_CC_HASH_MID")
+ self.assertIsNotNone(no_cc_profile)
+ self.assertIsNone(getattr(no_cc_profile, 'confirmation_code_hash', None),
+ "Profile should not have confirmation code hash for error test")
+
+ def tearDown(self):
+ """Clean up test environment."""
+ if hasattr(self, 'profile_store'):
+ self.profile_store.close()
+
+ if hasattr(self, 'temp_dir'):
+ import shutil
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+
+if __name__ == '__main__':
+ unittest.main()
\ No newline at end of file
diff --git a/tests/unittests/test_unified_9p_2p.py b/tests/unittests/test_unified_9p_2p.py
new file mode 100644
index 0000000..34c2eb9
--- /dev/null
+++ b/tests/unittests/test_unified_9p_2p.py
@@ -0,0 +1,194 @@
+#!/usr/bin/env python3
+# Test script to verify the unified ES2+/ES9+ architecture.
+#
+# (C) 2025 by Eric Wild <ewild(a)sysmocom.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import sys
+import logging
+import unittest
+import importlib.util
+from pathlib import Path
+
+# Add project root to path for imports
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+def import_osmo_smdpp():
+ project_root = Path(__file__).parent.parent.parent
+ module_path = project_root / 'osmo-smdpp.py'
+ if 'osmo_smdpp' in sys.modules:
+ return sys.modules['osmo_smdpp']
+
+ spec = importlib.util.spec_from_file_location('osmo_smdpp', module_path)
+ if spec is None or spec.loader is None:
+ raise ImportError(f"Cannot load module from {module_path}")
+
+ module = importlib.util.module_from_spec(spec)
+
+ # Register it in sys.modules with underscores so pickle will look for 'osmo_smdpp' not 'osmo-smdpp'
+ sys.modules['osmo_smdpp'] = module
+ spec.loader.exec_module(module)
+ return module
+
+osmo_smdpp = import_osmo_smdpp()
+
+SmDppHttpServer = osmo_smdpp.SmDppHttpServer
+Es2PlusProfileState = osmo_smdpp.Es2PlusProfileState
+
+
+class UnifiedArchitectureTest(unittest.TestCase):
+ """Test cases for unified ES2+/ES9+ architecture."""
+
+ def setUp(self):
+ """Set up test environment."""
+ logging.basicConfig(level=logging.DEBUG)
+ self.logger = logging.getLogger(__name__)
+
+ self.server = SmDppHttpServer(
+ server_hostname='test.example.com',
+ ci_certs_path=str(project_root / 'smdpp-data/certs/CertificateIssuer'),
+ common_cert_path=str(project_root / 'smdpp-data/certs'),
+ use_brainpool=False,
+ in_memory=True,
+ test_mode=True
+ )
+
+ def test_server_initialization(self):
+ """Test that server initializes correctly in test mode."""
+ self.assertIsNotNone(self.server)
+ self.assertTrue(self.server.test_mode)
+ self.assertIsNotNone(self.server.profile_store)
+
+ def test_static_profiles_loaded(self):
+ """Test that static test profiles are loaded."""
+ self.assertGreater(len(self.server.activation_code_profiles), 0,
+ "No static test profiles loaded")
+
+ def test_es2plus_profile_creation(self):
+ """Test creating and storing ES2+ profiles."""
+ test_matching_id = 'TEST123'
+ test_iccid = '8900000000000099999F'
+
+ es2_profile = Es2PlusProfileState(test_iccid, 'Test', 'TestOperator')
+ es2_profile.matching_id = test_matching_id
+ es2_profile.state = 'released'
+ self.server.profile_store[test_iccid] = es2_profile
+
+ self.assertIn(test_iccid, self.server.profile_store)
+ stored_profile = self.server.profile_store[test_iccid]
+ self.assertEqual(stored_profile.matching_id, test_matching_id)
+ self.assertEqual(stored_profile.state, 'released')
+
+ def test_es2plus_profile_lookup_by_matching_id(self):
+ """Test looking up ES2+ profiles by matching ID."""
+ test_matching_id = 'TEST_LOOKUP_001'
+ test_iccid = '8900000000000088888F'
+
+ profile = Es2PlusProfileState(test_iccid, 'Test', 'TestOperator')
+ profile.matching_id = test_matching_id
+ profile.state = 'confirmed'
+ self.server.profile_store[test_iccid] = profile
+
+ found_profile = self.server.profile_store.find_by_matching_id(test_matching_id)
+ self.assertIsNotNone(found_profile, "Profile not found by matching ID")
+ self.assertEqual(found_profile.iccid, test_iccid)
+ self.assertEqual(found_profile.matching_id, test_matching_id)
+
+ def test_profile_state_validation(self):
+ """Test that profile states are properly validated."""
+ # Test valid states
+ valid_states = ['released', 'confirmed']
+ for state in valid_states:
+ with self.subTest(state=state):
+ iccid = f'89000000000000{state[:5].upper()}F'
+ profile = Es2PlusProfileState(iccid, 'Test', 'TestOp')
+ profile.matching_id = f'TEST_{state.upper()}'
+ profile.state = state
+ self.server.profile_store[iccid] = profile
+
+ found = self.server.profile_store.find_by_matching_id(profile.matching_id)
+ self.assertIsNotNone(found)
+ self.assertEqual(found.state, state)
+
+ # Test invalid state (should still be stored but not ready for download)
+ invalid_state = 'allocated'
+ iccid = '8900000000000077777F'
+ profile = Es2PlusProfileState(iccid, 'Test', 'TestOp')
+ profile.matching_id = 'TEST_INVALID_STATE'
+ profile.state = invalid_state
+ self.server.profile_store[iccid] = profile
+
+ found = self.server.profile_store.find_by_matching_id('TEST_INVALID_STATE')
+ self.assertIsNotNone(found)
+ self.assertEqual(found.state, invalid_state)
+
+ def test_es2plus_takes_precedence_over_static(self):
+ """Test that ES2+ profiles take precedence over static profiles."""
+ # Find a matching ID that exists in static profiles
+ if not self.server.activation_code_profiles:
+ self.skipTest("No static profiles available for precedence test")
+
+ static_matching_id = next(iter(self.server.activation_code_profiles.keys()))
+ static_iccid = self.server.activation_code_profiles[static_matching_id]['iccid']
+
+ # Create ES2+ profile with same matching ID but different ICCID
+ es2_iccid = '8900000000000099999F'
+ es2_profile = Es2PlusProfileState(es2_iccid, 'Test', 'TestOperator')
+ es2_profile.matching_id = static_matching_id
+ es2_profile.state = 'released'
+ self.server.profile_store[es2_iccid] = es2_profile
+
+ # Verify ES2+ profile is found (takes precedence)
+ found_profile = self.server.profile_store.find_by_matching_id(static_matching_id)
+ self.assertIsNotNone(found_profile)
+ self.assertEqual(found_profile.iccid, es2_iccid,
+ "ES2+ profile should take precedence over static profile")
+ self.assertNotEqual(found_profile.iccid, static_iccid,
+ "Should return ES2+ ICCID, not static ICCID")
+
+ def test_multiple_profiles_coexistence(self):
+ """Test that multiple ES2+ profiles can coexist with different states."""
+ profiles_data = [
+ ('8900000000000011111F', 'TEST_MULTI_1', 'released'),
+ ('8900000000000022222F', 'TEST_MULTI_2', 'confirmed'),
+ ('8900000000000033333F', 'TEST_MULTI_3', 'allocated'),
+ ]
+
+ # Create and store multiple profiles
+ for iccid, matching_id, state in profiles_data:
+ profile = Es2PlusProfileState(iccid, 'Test', 'TestOperator')
+ profile.matching_id = matching_id
+ profile.state = state
+ self.server.profile_store[iccid] = profile
+
+ # Verify all profiles can be found by their matching IDs
+ for iccid, matching_id, state in profiles_data:
+ with self.subTest(matching_id=matching_id):
+ found = self.server.profile_store.find_by_matching_id(matching_id)
+ self.assertIsNotNone(found)
+ self.assertEqual(found.iccid, iccid)
+ self.assertEqual(found.state, state)
+
+ def tearDown(self):
+ """Clean up after tests."""
+ if hasattr(self, 'server') and self.server:
+ # Close profile store if it has a close method
+ if hasattr(self.server.profile_store, 'close'):
+ self.server.profile_store.close()
+
+
+if __name__ == '__main__':
+ unittest.main()
\ No newline at end of file
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/40883?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I8c7cf192695fdf5c37cd39effffcd9056085a2e5
Gerrit-Change-Number: 40883
Gerrit-PatchSet: 1
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
3 weeks, 5 days
- 1
- 0
[XS] Change in pysim[master]: x509 cert: fix weird cert check
by Hoernchen
Attention is currently required from: Hoernchen.
Hello Jenkins Builder, laforge,
I'd like you to reexamine a change. Please visit
https://gerrit.osmocom.org/c/pysim/+/40524?usp=email
to look at the new patch set (#8).
The following approvals got outdated and were removed:
Verified+1 by Jenkins Builder
The change is no longer submittable: Verified is unsatisfied now.
Change subject: x509 cert: fix weird cert check
......................................................................
x509 cert: fix weird cert check
Change-Id: I18beab0e1b24579724704c4141a2c457b2d4cf99
---
M pySim/esim/x509_cert.py
1 file changed, 1 insertion(+), 1 deletion(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/24/40524/8
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/40524?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newpatchset
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I18beab0e1b24579724704c4141a2c457b2d4cf99
Gerrit-Change-Number: 40524
Gerrit-PatchSet: 8
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
Gerrit-Reviewer: Jenkins Builder
Gerrit-Reviewer: laforge <laforge(a)osmocom.org>
Gerrit-Attention: Hoernchen <ewild(a)sysmocom.de>
3 weeks, 5 days
- 1
- 0
[S] Change in pysim[master]: smdpp: verify cert chain
by Hoernchen
Attention is currently required from: Hoernchen, dexter.
Hello Jenkins Builder, dexter, laforge,
I'd like you to reexamine a change. Please visit
https://gerrit.osmocom.org/c/pysim/+/40469?usp=email
to look at the new patch set (#13).
The following approvals got outdated and were removed:
Verified+1 by Jenkins Builder
The change is no longer submittable: Verified is unsatisfied now.
Change subject: smdpp: verify cert chain
......................................................................
smdpp: verify cert chain
Change-Id: I1e4e4b1b032dc6a8b7d15bd80d533a50fe0cff15
---
M osmo-smdpp.py
1 file changed, 20 insertions(+), 7 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/69/40469/13
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/40469?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newpatchset
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I1e4e4b1b032dc6a8b7d15bd80d533a50fe0cff15
Gerrit-Change-Number: 40469
Gerrit-PatchSet: 13
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
Gerrit-Reviewer: Jenkins Builder
Gerrit-Reviewer: dexter <pmaier(a)sysmocom.de>
Gerrit-Reviewer: laforge <laforge(a)osmocom.org>
Gerrit-Attention: Hoernchen <ewild(a)sysmocom.de>
Gerrit-Attention: dexter <pmaier(a)sysmocom.de>
3 weeks, 5 days
- 1
- 0
[M] Change in pysim[master]: smdpp: less verbose by default
by Hoernchen
Attention is currently required from: Hoernchen, dexter, laforge.
Hello Jenkins Builder, dexter, laforge,
I'd like you to reexamine a change. Please visit
https://gerrit.osmocom.org/c/pysim/+/40468?usp=email
to look at the new patch set (#12).
The following approvals got outdated and were removed:
Verified+1 by Jenkins Builder
Change subject: smdpp: less verbose by default
......................................................................
smdpp: less verbose by default
Those data blobs are huge.
Change-Id: I04a72b8f52417862d4dcba1f0743700dd942ef49
---
M osmo-smdpp.py
M pySim/esim/bsp.py
M pySim/esim/es8p.py
3 files changed, 86 insertions(+), 50 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/68/40468/12
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/40468?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newpatchset
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I04a72b8f52417862d4dcba1f0743700dd942ef49
Gerrit-Change-Number: 40468
Gerrit-PatchSet: 12
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
Gerrit-Reviewer: Jenkins Builder
Gerrit-Reviewer: dexter <pmaier(a)sysmocom.de>
Gerrit-Reviewer: laforge <laforge(a)osmocom.org>
Gerrit-CC: lynxis lazus <lynxis(a)fe80.eu>
Gerrit-Attention: Hoernchen <ewild(a)sysmocom.de>
Gerrit-Attention: laforge <laforge(a)osmocom.org>
Gerrit-Attention: dexter <pmaier(a)sysmocom.de>
3 weeks, 5 days
- 1
- 0
[M] Change in pysim[master]: smdpp: fix asn1tool OBJECT IDENTIFIER decoding
by Hoernchen
Attention is currently required from: Hoernchen, laforge.
Hello Jenkins Builder, laforge,
I'd like you to reexamine a change. Please visit
https://gerrit.osmocom.org/c/pysim/+/40526?usp=email
to look at the new patch set (#7).
The following approvals got outdated and were removed:
Code-Review+1 by laforge, Verified+1 by Jenkins Builder
Change subject: smdpp: fix asn1tool OBJECT IDENTIFIER decoding
......................................................................
smdpp: fix asn1tool OBJECT IDENTIFIER decoding
While at it make the linter happy.
The feature to ignore blocks is making slow progress:
https://github.com/astral-sh/ruff/issues/3711#
Change-Id: Ic678e6c4a4c1a01de87a8dce26f4a5e452e8562a
---
M osmo-smdpp.py
1 file changed, 113 insertions(+), 29 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/26/40526/7
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/40526?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: newpatchset
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: Ic678e6c4a4c1a01de87a8dce26f4a5e452e8562a
Gerrit-Change-Number: 40526
Gerrit-PatchSet: 7
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
Gerrit-Reviewer: Jenkins Builder
Gerrit-Reviewer: laforge <laforge(a)osmocom.org>
Gerrit-Attention: Hoernchen <ewild(a)sysmocom.de>
Gerrit-Attention: laforge <laforge(a)osmocom.org>
3 weeks, 5 days
- 1
- 0
[M] Change in pysim[master]: smdpp: less verbose by default
by lynxis lazus
Attention is currently required from: Hoernchen, dexter, laforge.
lynxis lazus has posted comments on this change by Hoernchen. ( https://gerrit.osmocom.org/c/pysim/+/40468?usp=email )
Change subject: smdpp: less verbose by default
......................................................................
Patch Set 11:
(1 comment)
File osmo-smdpp.py:
https://gerrit.osmocom.org/c/pysim/+/40468/comment/67edbff9_f2fccd0f?usp=em… :
PS6, Line 370: logger.debug("Tx JSON: %s" % json.dumps(output)[:200])
> Ok, but still not related to the change. NOTHING here is about performance.
Maybe it is not about the less than optimal performant. Ok, you could do it in a different commit.
But if you want to replace print() with logger.debug(), please use the logger.debug('foo %s', args) way, which is the intended way of use logger.debug() with such string substitude formats.
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/40468?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: comment
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I04a72b8f52417862d4dcba1f0743700dd942ef49
Gerrit-Change-Number: 40468
Gerrit-PatchSet: 11
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
Gerrit-Reviewer: Jenkins Builder
Gerrit-Reviewer: dexter <pmaier(a)sysmocom.de>
Gerrit-Reviewer: laforge <laforge(a)osmocom.org>
Gerrit-CC: lynxis lazus <lynxis(a)fe80.eu>
Gerrit-Attention: Hoernchen <ewild(a)sysmocom.de>
Gerrit-Attention: laforge <laforge(a)osmocom.org>
Gerrit-Attention: dexter <pmaier(a)sysmocom.de>
Gerrit-Comment-Date: Fri, 15 Aug 2025 10:25:55 +0000
Gerrit-HasComments: Yes
Gerrit-Has-Labels: No
Comment-In-Reply-To: Hoernchen <ewild(a)sysmocom.de>
Comment-In-Reply-To: laforge <laforge(a)osmocom.org>
Comment-In-Reply-To: lynxis lazus <lynxis(a)fe80.eu>
3 weeks, 5 days
- 1
- 0
[M] Change in pysim[master]: smdpp: less verbose by default
by Hoernchen
Attention is currently required from: dexter, laforge, lynxis lazus.
Hoernchen has posted comments on this change by Hoernchen. ( https://gerrit.osmocom.org/c/pysim/+/40468?usp=email )
Change subject: smdpp: less verbose by default
......................................................................
Patch Set 11:
(1 comment)
File osmo-smdpp.py:
https://gerrit.osmocom.org/c/pysim/+/40468/comment/45c5e6d7_ddd5e627?usp=em… :
PS6, Line 370: logger.debug("Tx JSON: %s" % json.dumps(output)[:200])
> 1) logger.debug("foo %s" % json.dumps(output)[:200]) will always do the string. […]
Ok, but still not related to the change. NOTHING here is about performance.
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/40468?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email
Gerrit-MessageType: comment
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I04a72b8f52417862d4dcba1f0743700dd942ef49
Gerrit-Change-Number: 40468
Gerrit-PatchSet: 11
Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>
Gerrit-Reviewer: Jenkins Builder
Gerrit-Reviewer: dexter <pmaier(a)sysmocom.de>
Gerrit-Reviewer: laforge <laforge(a)osmocom.org>
Gerrit-CC: lynxis lazus <lynxis(a)fe80.eu>
Gerrit-Attention: laforge <laforge(a)osmocom.org>
Gerrit-Attention: lynxis lazus <lynxis(a)fe80.eu>
Gerrit-Attention: dexter <pmaier(a)sysmocom.de>
Gerrit-Comment-Date: Thu, 14 Aug 2025 18:36:15 +0000
Gerrit-HasComments: Yes
Gerrit-Has-Labels: No
Comment-In-Reply-To: Hoernchen <ewild(a)sysmocom.de>
Comment-In-Reply-To: laforge <laforge(a)osmocom.org>
Comment-In-Reply-To: lynxis lazus <lynxis(a)fe80.eu>
3 weeks, 6 days
- 1
- 0