catarrh/pysim
1
0
Fork 0
mirror of https://gitea.osmocom.org/sim-card/pysim.git synced 2026-03-17 02:48:34 +03:00
Code Issues Packages Projects Releases Wiki Activity
Files
4e27b5107bbd222ce8cc54e14730317a659498d8
pysim/osmo-smdpp.py
Eric Wild 4e27b5107b smdpp: proper checks, proper check order, test mode 
Change-Id: Ie4db65594b2eaf6c95ffc5e73ff9ae61c4d9d3a3
2025-06-25 10:25:35 +02:00

1593 lines
78 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
# Early proof-of-concept towards a SM-DP+ HTTP service for GSMA consumer eSIM RSP
#
# (C) 2023-2024 by Harald Welte <laforge@osmocom.org>
#
# 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/>.
# asn1tools issue https://github.com/eerimoq/asn1tools/issues/194
# must be first here
import asn1tools
import asn1tools.codecs.ber
import asn1tools.codecs.der
# do not move the code
def fix_asn1_oid_decoding():
fix_asn1_schema = """
TestModule DEFINITIONS ::= BEGIN
TestOid ::= SEQUENCE {
oid OBJECT IDENTIFIER
}
END
"""
fix_asn1_asn1 = asn1tools.compile_string(fix_asn1_schema, codec='der')
fix_asn1_oid_string = '2.999.10'
fix_asn1_encoded = fix_asn1_asn1.encode('TestOid', {'oid': fix_asn1_oid_string})
fix_asn1_decoded = fix_asn1_asn1.decode('TestOid', fix_asn1_encoded)
if (fix_asn1_decoded['oid'] != fix_asn1_oid_string):
# ASN.1 OBJECT IDENTIFIER Decoding Issue:
#
# In ASN.1 BER/DER encoding, the first two arcs of an OBJECT IDENTIFIER are
# combined into a single value: (40 * arc0) + arc1. This is encoded as a base-128
# variable-length quantity (and commonly known as VLQ or base-128 encoding)
# as specified in ITU-T X.690 §8.19, it can span multiple bytes if
# the value is large.
#
# For arc0 = 0 or 1, arc1 must be in [0, 39]. For arc0 = 2, arc1 can be any non-negative integer.
# All subsequent arcs (arc2, arc3, ...) are each encoded as a separate base-128 VLQ.
#
# The decoding bug occurs when the decoder does not properly split the first
# subidentifier for arc0 = 2 and arc1 >= 40. Instead of decoding:
# - arc0 = 2
# - arc1 = (first_subidentifier - 80)
# it may incorrectly interpret the first_subidentifier as arc0 = (first_subidentifier // 40),
# arc1 = (first_subidentifier % 40), which is only valid for arc1 < 40.
#
# This patch handles it properly for all valid OBJECT IDENTIFIERs
# with large second arcs, by applying the ASN.1 rules:
# - if first_subidentifier < 40: arc0 = 0, arc1 = first_subidentifier
# - elif first_subidentifier < 80: arc0 = 1, arc1 = first_subidentifier - 40
# - else: arc0 = 2, arc1 = first_subidentifier - 80
#
# This problem is not uncommon, see for example https://github.com/randombit/botan/issues/4023
def fixed_decode_object_identifier(data, offset, end_offset):
"""Decode ASN.1 OBJECT IDENTIFIER from bytes to dotted string, fixing large second arc handling."""
def read_subidentifier(data, offset):
value = 0
while True:
b = data[offset]
value = (value << 7) | (b & 0x7F)
offset += 1
if not (b & 0x80):
break
return value, offset
subid, offset = read_subidentifier(data, offset)
if subid < 40:
first = 0
second = subid
elif subid < 80:
first = 1
second = subid - 40
else:
first = 2
second = subid - 80
arcs = [first, second]
while offset < end_offset:
subid, offset = read_subidentifier(data, offset)
arcs.append(subid)
return '.'.join(str(x) for x in arcs)
asn1tools.codecs.ber.decode_object_identifier = fixed_decode_object_identifier
asn1tools.codecs.der.decode_object_identifier = fixed_decode_object_identifier
# test our patch
asn1 = asn1tools.compile_string(fix_asn1_schema, codec='der')
decoded = asn1.decode('TestOid', fix_asn1_encoded)['oid']
assert fix_asn1_oid_string == str(decoded)
fix_asn1_oid_decoding()
from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature, encode_dss_signature
from cryptography import x509
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec, dh
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat, PrivateFormat, NoEncryption, ParameterFormat
from cryptography.x509.oid import ExtensionOID, NameOID
from pathlib import Path
import json
import sys
import argparse
import uuid
import os
import functools
from typing import Optional, Dict, List
from pprint import pprint as pp
import base64
import time
from base64 import b64decode
from klein import Klein
from twisted.web.iweb import IRequest
from osmocom.utils import h2b, b2h, swap_nibbles
import pySim.esim.rsp as rsp
from pySim.esim import saip, PMO
from pySim.esim.es8p import *
from pySim.esim.x509_cert import oid, cert_policy_has_oid, cert_get_auth_key_id
from cryptography.x509 import ExtensionNotFound
from pySim.esim.x509_cert import CertAndPrivkey, CertificateSet, cert_get_subject_key_id
from pySim.esim import x509_err
from datetime import datetime, timezone
import hashlib
import logging
logger = logging.getLogger(__name__)
# HACK: make this configurable
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')
def set_headers(request: IRequest):
"""Set the request headers as mandatory by GSMA eSIM RSP."""
request.setHeader('Content-Type', 'application/json;charset=UTF-8')
request.setHeader('X-Admin-Protocol', 'gsma/rsp/v2.1.0')
def validate_request_headers(request: IRequest):
"""Validate mandatory HTTP headers according to SGP.22."""
content_type = request.getHeader('Content-Type')
if not content_type or not content_type.startswith('application/json'):
raise ApiError('1.2.1', '2.1', 'Invalid Content-Type header')
admin_protocol = request.getHeader('X-Admin-Protocol')
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
#print("Using nameConstraints parsing for variant O certificate")
permitted_iins.extend(_parse_name_constraints_eins(eum_cert))
else:
# New variants (Ov3, A, B, C) - use GSMA permittedEins extension
#print("Using GSMA permittedEins parsing for newer certificate variant")
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
)
# print("Found nameConstraints extension (variant O)")
name_constraints = name_constraints_ext.value
# Check permittedSubtrees for IIN constraints
if name_constraints.permitted_subtrees:
for subtree in name_constraints.permitted_subtrees:
# print(f"Processing permitted subtree: {subtree}")
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
r = int.from_bytes(sig[0:32], 'big')
s = int.from_bytes(sig[32:32*2], 'big')
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):
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()
@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
instances."""
certs = []
for dirpath, dirnames, filenames in os.walk(path):
for filename in filenames:
cert = None
if filename.endswith('.der'):
with open(os.path.join(dirpath, filename), 'rb') as f:
cert = x509.load_der_x509_certificate(f.read())
elif filename.endswith('.pem'):
with open(os.path.join(dirpath, filename), 'rb') as f:
cert = x509.load_pem_x509_certificate(f.read())
if cert:
# verify it is a CI certificate (keyCertSign + i-rspRole-ci)
if not cert_policy_has_oid(cert, oid.id_rspRole_ci):
raise ValueError("alleged CI certificate %s doesn't have CI policy" % filename)
certs.append(cert)
return certs
def ci_get_cert_for_pkid(self, ci_pkid: bytes) -> Optional[x509.Certificate]:
"""Find CI certificate for given key identifier."""
for cert in self.ci_certs:
logger.debug("cert: %s" % cert)
subject_exts = list(filter(lambda x: isinstance(x.value, x509.SubjectKeyIdentifier), cert.extensions))
logger.debug(subject_exts)
subject_pkid = subject_exts[0].value
logger.debug(subject_pkid)
if subject_pkid and subject_pkid.key_identifier == ci_pkid:
return cert
return None
def validate_certificate_chain_for_verification(self, euicc_ci_pkid_list: List[bytes]) -> bool:
"""Validate that SM-DP+ has valid certificate chains for the given CI PKIDs."""
for ci_pkid in euicc_ci_pkid_list:
ci_cert = self.ci_get_cert_for_pkid(ci_pkid)
if ci_cert:
# Check if our DPauth certificate chains to this CI
try:
cs = CertificateSet(ci_cert)
cs.verify_cert_chain(self.dp_auth.cert)
return True
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, 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 {}
self.require_confirmation_code = False # Global flag - disabled by default
# load DPauth cert + key
self.dp_auth = CertAndPrivkey(oid.id_rspRole_dp_auth_v2)
cert_dir = common_cert_path
if use_brainpool:
self.dp_auth.cert_from_der_file(os.path.join(cert_dir, 'DPauth', 'CERT_S_SM_DPauth_ECDSA_BRP.der'))
self.dp_auth.privkey_from_pem_file(os.path.join(cert_dir, 'DPauth', 'SK_S_SM_DPauth_ECDSA_BRP.pem'))
else:
self.dp_auth.cert_from_der_file(os.path.join(cert_dir, 'DPauth', 'CERT_S_SM_DPauth_ECDSA_NIST.der'))
self.dp_auth.privkey_from_pem_file(os.path.join(cert_dir, 'DPauth', 'SK_S_SM_DPauth_ECDSA_NIST.pem'))
# load DPpb cert + key
self.dp_pb = CertAndPrivkey(oid.id_rspRole_dp_pb_v2)
if use_brainpool:
self.dp_pb.cert_from_der_file(os.path.join(cert_dir, 'DPpb', 'CERT_S_SM_DPpb_ECDSA_BRP.der'))
self.dp_pb.privkey_from_pem_file(os.path.join(cert_dir, 'DPpb', 'SK_S_SM_DPpb_ECDSA_BRP.pem'))
else:
self.dp_pb.cert_from_der_file(os.path.join(cert_dir, 'DPpb', 'CERT_S_SM_DPpb_ECDSA_NIST.der'))
self.dp_pb.privkey_from_pem_file(os.path.join(cert_dir, 'DPpb', 'SK_S_SM_DPpb_ECDSA_NIST.pem'))
if in_memory:
self.rss = rsp.RspSessionStore(in_memory=True)
logger.info("Using in-memory session storage")
else:
# Use different session database files for BRP and NIST to avoid file locking during concurrent runs
session_db_suffix = "BRP" if use_brainpool else "NIST"
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.used_euicc_challenges = set() # Track used eUICC challenges to prevent reuse
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):
request.setResponseCode(200)
pp(failure)
return failure.value.encode()
@staticmethod
def _ecdsa_verify(cert: x509.Certificate, signature: bytes, data: bytes) -> bool:
pubkey = cert.public_key()
dss_sig = ecdsa_tr03111_to_dss(signature)
try:
pubkey.verify(dss_sig, data, ec.ECDSA(hashes.SHA256()))
return True
except InvalidSignature:
return False
@staticmethod
def rsp_api_wrapper(func):
"""Wrapper that can be used as decorator in order to perform common REST API endpoint entry/exit
functionality, such as JSON decoding/encoding and debug-printing."""
@functools.wraps(func)
def _api_wrapper(self, request: IRequest):
validate_request_headers(request)
content = json.loads(request.content.read())
#logger.debug("Rx JSON: %s" % json.dumps(content))
set_headers(request)
output = func(self, request, content)
if output == None:
return ''
build_resp_header(output)
logger.debug("Tx JSON: %s" % json.dumps(output)[:200])
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:
"""See ES9+ InitiateAuthentication SGP.22 Section 5.6.1"""
# 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:
raise ValueError
# Check for eUICC challenge reuse (security requirement, not in spec)
if euiccChallenge in self.used_euicc_challenges:
raise ApiError('8.1', '3.1', 'eUICC challenge reuse detected') # Generic security error
self.used_euicc_challenges.add(euiccChallenge)
euiccInfo1_bin = b64decode(content['euiccInfo1'])
euiccInfo1 = rsp.asn1.decode('EUICCInfo1', euiccInfo1_bin)
logger.debug("Rx euiccInfo1: %s" % euiccInfo1)
# 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 ...
pkid_list = euiccInfo1['euiccCiPKIdListForSigning']
if 'euiccCiPKIdListForSigningV3' in euiccInfo1:
pkid_list = pkid_list + euiccInfo1['euiccCiPKIdListForSigningV3']
# 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.
try:
ci_subject_key_id = cert_get_subject_key_id(ci_cert)
except:
# 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()
assert not transactionId in self.rss
# Generate a serverChallenge for eUICC authentication attached to the ongoing RSP session.
serverChallenge = os.urandom(16)
# Generate a serverSigned1 data object as expected by the eUICC and described in section 5.7.13 "ES10b.AuthenticateServer". If and only if both eUICC and LPA indicate crlStaplingV3Support, the SM-DP+ SHALL indicate crlStaplingV3Used in sessionContext.
serverSigned1 = {
'transactionId': h2b(transactionId),
'euiccChallenge': euiccChallenge,
'serverAddress': self.server_hostname,
'serverChallenge': serverChallenge,
}
logger.debug("Tx serverSigned1: %s" % serverSigned1)
serverSigned1_bin = rsp.asn1.encode('ServerSigned1', serverSigned1)
logger.debug("Tx serverSigned1: %s" % rsp.asn1.decode('ServerSigned1', serverSigned1_bin))
output = {}
output['serverSigned1'] = b64encode2str(serverSigned1_bin)
# Generate a signature (serverSignature1) as described in section 5.7.13 "ES10b.AuthenticateServer" using the SK related to the selected CERT.DPauth.SIG.
# serverSignature1 SHALL be created using the private key associated to the RSP Server Certificate for authentication, and verified by the eUICC using the contained public key as described in section 2.6.9. serverSignature1 SHALL apply on serverSigned1 data object.
output['serverSignature1'] = b64encode2str(b'\x5f\x37\x40' + self.dp_auth.ecdsa_sign(serverSigned1_bin))
output['transactionId'] = transactionId
server_cert_aki = self.dp_auth.get_authority_key_identifier()
output['euiccCiPKIdToBeUsed'] = b64encode2str(b'\x04\x14' + server_cert_aki.key_identifier)
output['serverCertificate'] = b64encode2str(self.dp_auth.get_cert_as_der()) # CERT.DPauth.SIG
# FIXME: add those certificate
#output['otherCertsInChain'] = b64encode2str()
# create SessionState and store it in rss
self.rss[transactionId] = rsp.RspSessionState(transactionId, serverChallenge,
cert_get_subject_key_id(ci_cert))
return output
@app.route('/gsma/rsp2/es9plus/authenticateClient', methods=['POST'])
@rsp_api_wrapper
def authenticateClient(self, request: IRequest, content: dict) -> dict:
"""See ES9+ AuthenticateClient in SGP.22 Section 5.6.3"""
transactionId = content['transactionId']
authenticateServerResp_bin = b64decode(content['authenticateServerResponse'])
authenticateServerResp = rsp.asn1.decode('AuthenticateServerResponse', authenticateServerResp_bin)
logger.debug("Rx %s: %s" % authenticateServerResp)
if authenticateServerResp[0] == 'authenticateResponseError':
r_err = authenticateServerResp[1]
#r_err['transactionId']
#r_err['authenticateErrorCode']
raise ValueError("authenticateResponseError %s" % r_err)
r_ok = authenticateServerResp[1]
euiccSigned1 = r_ok['euiccSigned1']
euiccSigned1_bin = rsp.extract_euiccSigned1(authenticateServerResp_bin)
euiccSignature1_bin = r_ok['euiccSignature1']
euiccCertificate_dec = r_ok['euiccCertificate']
# TODO: use original data, don't re-encode?
euiccCertificate_bin = rsp.asn1.encode('Certificate', euiccCertificate_dec)
eumCertificate_dec = r_ok['eumCertificate']
eumCertificate_bin = rsp.asn1.encode('Certificate', eumCertificate_dec)
# TODO v3: otherCertsInChain
# load certificate
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.
ss = self.rss.get(transactionId, None)
if ss is None:
# 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?
# 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
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 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:
# Determine which certificate is revoked based on the error details
if 'EUM' in str(e) or (hasattr(e, 'cert') and e.cert == eum_cert):
# SGP.22 Table 49: EUM Certificate - Expired (revoked uses expired code)
raise ApiError('8.1.2', '6.3', 'Certificate has expired')
else:
# 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 as e:
# 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.
if not self._ecdsa_verify(euicc_cert, euiccSignature1_bin, euiccSigned1_bin):
# 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):
# 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.
if euiccSigned1['serverChallenge'] != ss.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 [...]
if euiccSigned1['ctxParams1'][0] == 'ctxParamsForCommonAuthentication':
cpca = euiccSigned1['ctxParams1'][1]
matchingId = cpca.get('matchingId', None)
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:
# Try to load from file system (legacy support)
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
# Create a 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'):
# Simple check - in 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 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')
# 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=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': 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)
# update non-volatile state with updated ss object
self.rss[transactionId] = ss
return {
'transactionId': transactionId,
'profileMetadata': b64encode2str(profileMetadata_bin),
'smdpSigned2': b64encode2str(smdpSigned2_bin),
'smdpSignature2': b64encode2str(ss.smdpSignature2_do),
'smdpCertificate': b64encode2str(self.dp_pb.get_cert_as_der()), # CERT.DPpb.SIG
}
@app.route('/gsma/rsp2/es9plus/getBoundProfilePackage', methods=['POST'])
@rsp_api_wrapper
def getBoundProfilePackage(self, request: IRequest, content: dict) -> dict:
"""See ES9+ GetBoundProfilePackage SGP.22 Section 5.6.2"""
transactionId = content['transactionId']
# 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'])
prepDownloadResp = rsp.asn1.decode('PrepareDownloadResponse', prepDownloadResp_bin)
logger.debug("Rx %s: %s" % prepDownloadResp)
if prepDownloadResp[0] == 'downloadResponseError':
r_err = prepDownloadResp[1]
#r_err['transactionId']
#r_err['downloadErrorCode']
# Download error from eUICC - use generic eUICC verification failed
download_error_code = r_err.get('downloadErrorCode', 'unknown')
# SGP.22 Table 51: eUICC - Verification Failed
raise ApiError('8.1', '6.1', 'eUICC signature is invalid')
r_ok = prepDownloadResp[1]
# Verify the euiccSignature2 computed over euiccSigned2 and smdpSignature2 using the PK.EUICC.SIG attached to the ongoing RSP session
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']:
# 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']
logger.debug("euiccOtpk: %s" % (b2h(ss.euicc_otpk)))
# 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())
# 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'
# Generate Session Keys using the CRT, otPK.eUICC.ECKA and otSK.DP.ECKA according to annex G
euicc_public_key = ec.EllipticCurvePublicKey.from_encoded_point(ss.smdp_ot.curve, ss.euicc_otpk)
ss.shared_secret = ss.smdp_ot.exchange(ec.ECDH(), euicc_public_key)
logger.debug("shared_secret: %s" % b2h(ss.shared_secret))
# 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')
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
# cluttering the log with stuff happening after the failure
#upp = UnprotectedProfilePackage.from_der(b'', metadata=ss.profileMetadata)
if False:
# Use random keys
bpp = BoundProfilePackage.from_upp(upp)
else:
# Use sesssion keys
ppp = ProtectedProfilePackage.from_upp(upp, BspInstance(b'\x00'*16, b'\x11'*16, b'\x22'*16))
bpp = BoundProfilePackage.from_ppp(ppp)
# update non-volatile state with updated ss object
self.rss[transactionId] = ss
return {
'transactionId': transactionId,
'boundProfilePackage': b64encode2str(bpp.encode(ss, self.dp_pb)),
}
@app.route('/gsma/rsp2/es9plus/handleNotification', methods=['POST'])
@rsp_api_wrapper
def handleNotification(self, request: IRequest, content: dict) -> dict:
"""See ES9+ HandleNotification in SGP.22 Section 5.6.4"""
# SGP.22 Section 6.3: "A normal notification function execution status (MEP Notification)
# SHALL be indicated by the HTTP status code '204' (No Content) with an empty HTTP response body"
request.setResponseCode(204)
pendingNotification_bin = b64decode(content['pendingNotification'])
pendingNotification = rsp.asn1.decode('PendingNotification', pendingNotification_bin)
logger.debug("Rx %s: %s" % pendingNotification)
if pendingNotification[0] == 'profileInstallationResult':
profileInstallRes = pendingNotification[1]
pird = profileInstallRes['profileInstallationResultData']
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']
# TODO: use original data, don't re-encode?
pird_bin = rsp.asn1.encode('ProfileInstallationResultData', pird)
# verify eUICC signature
if not self._ecdsa_verify(ss.euicc_cert, profileInstallRes['euiccSignPIR'], pird_bin):
# 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]
elif pendingNotification[0] == 'otherSignedNotification':
otherSignedNotif = pendingNotification[1]
# TODO: use some kind of partially-parsed original data, don't re-encode?
euiccCertificate_bin = rsp.asn1.encode('Certificate', otherSignedNotif['euiccCertificate'])
eumCertificate_bin = rsp.asn1.encode('Certificate', otherSignedNotif['eumCertificate'])
euicc_cert = x509.load_der_x509_certificate(euiccCertificate_bin)
eum_cert = x509.load_der_x509_certificate(eumCertificate_bin)
ci_cert_id = cert_get_auth_key_id(eum_cert)
# Verify the validity of the eUICC certificate chain
cs = CertificateSet(self.ci_get_cert_for_pkid(ci_cert_id))
cs.add_intermediate_cert(eum_cert)
# TODO v3: otherCertsInChain
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):
# 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
iccid = other_notif.get('iccid', None)
if iccid:
iccid = swap_nibbles(b2h(iccid))
logger.debug("handleNotification: EID %s: %s of %s" % (eid, pmo, iccid))
else:
# 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
#@app.route('/gsma/rsp3/es9plus/handleDeviceChangeRequest, methods=['POST']')
#@rsp_api_wrapper
#"""See ES9+ ConfirmDeviceChange in SGP.22 Section 5.6.6"""
# TODO: implement this
@app.route('/gsma/rsp2/es9plus/cancelSession', methods=['POST'])
@rsp_api_wrapper
def cancelSession(self, request: IRequest, content: dict) -> dict:
"""See ES9+ CancelSession in SGP.22 Section 5.6.5"""
logger.debug("Rx JSON: %s" % content)
transactionId = content['transactionId']
# Verify that the received transactionId is known and relates to an ongoing RSP session
ss = self.rss.get(transactionId, None)
if ss is None:
# 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)
logger.debug("Rx %s: %s" % cancelSessionResponse)
if cancelSessionResponse[0] == 'cancelSessionResponseError':
# FIXME: print some error
return
cancelSessionResponseOk = cancelSessionResponse[1]
# TODO: use original data, don't re-encode?
ecsr = cancelSessionResponseOk['euiccCancelSessionSigned']
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()
# 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):
# 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.
# TODO: 3. If required, execute the SM-DS Event Deletion procedure described in section 3.6.3.
# delete actual session data
del self.rss[transactionId]
# Per SGP.22 section 6.5.2.10, cancelSession returns an empty response (header only)
return {}
def main(argv):
parser = argparse.ArgumentParser()
parser.add_argument("-H", "--host", help="Host/IP to bind HTTP to", default="localhost")
parser.add_argument("-p", "--port", help="TCP port to bind HTTP to", default=8000)
parser.add_argument("-c", "--certdir", help=f"cert subdir relative to {DATA_DIR}", default="certs")
parser.add_argument("-s", "--nossl", help="do NOT use ssl", action='store_true', default=False)
parser.add_argument("-v", "--verbose", help="dump more raw info", action='store_true', default=False)
parser.add_argument("-b", "--brainpool", help="Use Brainpool curves instead of NIST",
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, test_mode=args.test)
if(args.nossl):
hs.app.run(args.host, args.port)
else:
curve_type = 'BRP' if args.brainpool else 'NIST'
cert_derpath = Path(common_cert_path) / 'DPtls' / f'CERT_S_SM_DP_TLS_{curve_type}.der'
cert_pempath = Path(common_cert_path) / 'DPtls' / f'CERT_S_SM_DP_TLS_{curve_type}.pem'
cert_skpath = Path(common_cert_path) / 'DPtls' / f'SK_S_SM_DP_TLS_{curve_type}.pem'
dhparam_path = Path(common_cert_path) / "dhparam2048.pem"
if not dhparam_path.exists():
print("Generating dh params, this takes a few seconds..")
# Generate DH parameters with 2048-bit key size and generator 2
parameters = dh.generate_parameters(generator=2, key_size=2048)
pem_data = parameters.parameter_bytes(encoding=Encoding.PEM,format=ParameterFormat.PKCS3)
with open(dhparam_path, 'wb') as file:
file.write(pem_data)
print("DH params created successfully")
if not cert_pempath.exists():
print("Translating tls server cert from DER to PEM..")
with open(cert_derpath, 'rb') as der_file:
der_cert_data = der_file.read()
cert = x509.load_der_x509_certificate(der_cert_data)
pem_cert = cert.public_bytes(Encoding.PEM) #.decode('utf-8')
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)
if __name__ == "__main__":
main(sys.argv)
Reference in New Issue View Git Blame Copy Permalink