June 2025 - gerrit-log - lists.osmocom.org

[XS] Change in osmo-ttcn3-hacks[master]: simtrace emulation: check that port is connected to prevent errors du...

by osmith

Attention is currently required from: Hoernchen. osmith has uploaded a new patch set (#2) to the change originally created by Hoernchen. ( https://gerrit.osmocom.org/c/osmo-ttcn3-hacks/+/27342?usp=email ) 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: simtrace emulation: check that port is connected to prevent errors due to delayed msg ...................................................................... simtrace emulation: check that port is connected to prevent errors due to delayed msg Change-Id: I5d5dda0cf1443223b2615f1059cfdcf698f1acae --- M library/SIMTRACE_Emulation.ttcn 1 file changed, 6 insertions(+), 2 deletions(-) git pull ssh://gerrit.osmocom.org:29418/osmo-ttcn3-hacks refs/changes/42/27342/2 -- To view, visit https://gerrit.osmocom.org/c/osmo-ttcn3-hacks/+/27342?usp=email To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email Gerrit-MessageType: newpatchset Gerrit-Project: osmo-ttcn3-hacks Gerrit-Branch: master Gerrit-Change-Id: I5d5dda0cf1443223b2615f1059cfdcf698f1acae Gerrit-Change-Number: 27342 Gerrit-PatchSet: 2 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>

2 months, 2 weeks

1
0
0 0

[L] Change in osmo-ttcn3-hacks[master]: octsim: initial commit

by osmith

Attention is currently required from: Hoernchen. osmith has uploaded a new patch set (#2) to the change originally created by Hoernchen. ( https://gerrit.osmocom.org/c/osmo-ttcn3-hacks/+/27341?usp=email ) The following approvals got outdated and were removed: Verified-1 by Jenkins Builder Change subject: octsim: initial commit ...................................................................... octsim: initial commit This testsuite is for the sysmoOCTSIM. Related: https://www.sysmocom.de/products/sim/sysmooctsim/ Tweaked-by: Eric Wild <ewild(a)sysmocom.de> Tweaked-by: Oliver Smith <osmith(a)sysmocom.de> Change-Id: Iada6422d694eb5fc862477c8b43b8642c8d96692 --- M Makefile A octsim/OCTSIM_Tests.cfg A octsim/OCTSIM_Tests.default A octsim/OCTSIM_Tests.ttcn A octsim/README.md A octsim/gen_links.sh A octsim/regen_makefile.sh 7 files changed, 313 insertions(+), 0 deletions(-) git pull ssh://gerrit.osmocom.org:29418/osmo-ttcn3-hacks refs/changes/41/27341/2 -- To view, visit https://gerrit.osmocom.org/c/osmo-ttcn3-hacks/+/27341?usp=email To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email Gerrit-MessageType: newpatchset Gerrit-Project: osmo-ttcn3-hacks Gerrit-Branch: master Gerrit-Change-Id: Iada6422d694eb5fc862477c8b43b8642c8d96692 Gerrit-Change-Number: 27341 Gerrit-PatchSet: 2 Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de> Gerrit-Reviewer: Jenkins Builder Gerrit-CC: laforge <laforge(a)osmocom.org> Gerrit-Attention: Hoernchen <ewild(a)sysmocom.de>

2 months, 2 weeks

1
0
0 0

[XS] Change in osmo-ttcn3-hacks[master]: simtrace emulation: set usb configuration

by osmith

Attention is currently required from: Hoernchen, laforge. osmith has uploaded a new patch set (#2) to the change originally created by Hoernchen. ( https://gerrit.osmocom.org/c/osmo-ttcn3-hacks/+/27343?usp=email ) The following approvals got outdated and were removed: Code-Review+1 by laforge, Verified-1 by Jenkins Builder Change subject: simtrace emulation: set usb configuration ...................................................................... simtrace emulation: set usb configuration Tweaked-by: Oliver Smith <osmith(a)sysmocom.de> Change-Id: I9def79343b4d2575f8b1bc7b9c9c8170b631ed49 --- M deps/Makefile M library/SIMTRACE_Emulation.ttcn 2 files changed, 2 insertions(+), 1 deletion(-) git pull ssh://gerrit.osmocom.org:29418/osmo-ttcn3-hacks refs/changes/43/27343/2 -- To view, visit https://gerrit.osmocom.org/c/osmo-ttcn3-hacks/+/27343?usp=email To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email Gerrit-MessageType: newpatchset Gerrit-Project: osmo-ttcn3-hacks Gerrit-Branch: master Gerrit-Change-Id: I9def79343b4d2575f8b1bc7b9c9c8170b631ed49 Gerrit-Change-Number: 27343 Gerrit-PatchSet: 2 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>

2 months, 2 weeks

1
0
0 0

[L] Change in pysim[master]: smdpp: add proper tls support, cert generation FOR TESTING

by Hoernchen

Attention is currently required from: dexter, laforge. Hoernchen has posted comments on this change by Hoernchen. ( https://gerrit.osmocom.org/c/pysim/+/40464?usp=email ) Change subject: smdpp: add proper tls support, cert generation FOR TESTING ...................................................................... Patch Set 6: (7 comments) File contrib/generate_smdpp_certs.py: https://gerrit.osmocom.org/c/pysim/+/40464/comment/339f62a9_c8663cc1?usp=em… : PS5, Line 2: > To me this looks like an independent tool. […] no, this is the tool and these are the options, these things belong together and make absolutely no sense as single commits with random mystery options and useless certs. File osmo-smdpp.py: https://gerrit.osmocom.org/c/pysim/+/40464/comment/4382ca7d_ea805b69?usp=em… : PS5, Line 26: from pathlib import Path > You have moved a lot of those imports from another code location. […] Done https://gerrit.osmocom.org/c/pysim/+/40464/comment/bc8c1331_3407f1b1?usp=em… : PS5, Line 53: > We now support SSL and we have means to point to different certificate sub directories. […] not related to this patch at all, we cann add stuff if we need it later. https://gerrit.osmocom.org/c/pysim/+/40464/comment/6ee98918_c7b38b8d?usp=em… : PS5, Line 586: parser.add_argument("-p", "--port", help="TCP port to bind HTTP to", default=8000) > Maybe 8443, since we now support SSL? no, because that is being used by something else on my system, we really do not want to default to a port commonly used for anything like https for a toy smdpp. https://gerrit.osmocom.org/c/pysim/+/40464/comment/3817ef0e_da2c40d3?usp=em… : PS5, Line 587: parser.add_argument("-c", "--certpath", help=f"cert subdir relative to {DATA_DIR}", default="certs") > I think --certpath is a bit irritating as one may thing that it is possible to provide an absolute p […] Done .. the help line literally explains that it is relative https://gerrit.osmocom.org/c/pysim/+/40464/comment/a7b3dcc9_fe1f3390?usp=em… : PS5, Line 588: parser.add_argument("-s", "--nossl", help="do NOT use ssl", action='store_true', default=False) > I wonder if this may break anything when osmo-smdpp suddenly supports SSL by default. […] The number of users is probably... low.. anyway. https://gerrit.osmocom.org/c/pysim/+/40464/comment/f5aa1340_7306aee3?usp=em… : PS5, Line 591: common_cert_path = os.path.join(DATA_DIR, args.certpath) > (see above) Maybe having DATA_DIR as a commandline parameter would be a good idea. I dont't need it, and it is not related to this patch either. -- To view, visit https://gerrit.osmocom.org/c/pysim/+/40464?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: I84b2666422b8ff565620f3827ef4d4d7635a21be Gerrit-Change-Number: 40464 Gerrit-PatchSet: 6 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: laforge <laforge(a)osmocom.org> Gerrit-Attention: dexter <pmaier(a)sysmocom.de> Gerrit-Comment-Date: Wed, 25 Jun 2025 08:51:06 +0000 Gerrit-HasComments: Yes Gerrit-Has-Labels: No Comment-In-Reply-To: dexter <pmaier(a)sysmocom.de>

2 months, 2 weeks

1
0
0 0

[XS] Change in pysim[master]: x509 cert: fix weird cert check

by Hoernchen

Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/40524?usp=email ) 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/1 diff --git a/pySim/esim/x509_cert.py b/pySim/esim/x509_cert.py index c8a0386..8139a09 100644 --- a/pySim/esim/x509_cert.py +++ b/pySim/esim/x509_cert.py @@ -149,7 +149,7 @@ check_signed(c, self.root_cert) return parent_cert = self.intermediate_certs.get(aki, None) - if not aki: + if not parent_cert: raise VerifyError('Could not find intermediate certificate for AuthKeyId %s' % b2h(aki)) check_signed(c, parent_cert) # if we reach here, we passed (no exception raised) -- 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: newchange Gerrit-Project: pysim Gerrit-Branch: master Gerrit-Change-Id: I18beab0e1b24579724704c4141a2c457b2d4cf99 Gerrit-Change-Number: 40524 Gerrit-PatchSet: 1 Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>

2 months, 2 weeks

1
0
0 0

[S] Change in pysim[master]: smdpp: add proper brp cert support

by Hoernchen

Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/40525?usp=email ) Change subject: smdpp: add proper brp cert support ...................................................................... smdpp: add proper brp cert support Change-Id: I6906732f7d193a9c2234075f4a82df5e0ed46100 --- M .gitignore M osmo-smdpp.py 2 files changed, 12 insertions(+), 6 deletions(-) git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/25/40525/1 diff --git a/.gitignore b/.gitignore index 6b74841..7a2e150 100644 --- a/.gitignore +++ b/.gitignore @@ -7,9 +7,10 @@ /.local /build /pySim.egg-info -/smdpp-data/sm-dp-sessions +/smdpp-data/sm-dp-sessions* dist tags smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.pem +smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.pem smdpp-data/generated smdpp-data/certs/dhparam2048.pem diff --git a/osmo-smdpp.py b/osmo-smdpp.py index b3494ec..41c3562 100755 --- a/osmo-smdpp.py +++ b/osmo-smdpp.py @@ -346,7 +346,9 @@ 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')) - self.rss = rsp.RspSessionStore(os.path.join(DATA_DIR, "sm-dp-sessions")) + # 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" + self.rss = rsp.RspSessionStore(os.path.join(DATA_DIR, f"sm-dp-sessions-{session_db_suffix}")) @app.handle_errors(ApiError) def handle_apierror(self, request: IRequest, failure): @@ -777,18 +779,21 @@ 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) 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=False) + 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) if(args.nossl): hs.app.run(args.host, args.port) else: - cert_derpath = Path(common_cert_path) / 'DPtls' / 'CERT_S_SM_DP_TLS_NIST.der' - cert_pempath = Path(common_cert_path) / 'DPtls' / 'CERT_S_SM_DP_TLS_NIST.pem' - cert_skpath = Path(common_cert_path) / 'DPtls' / 'SK_S_SM_DP_TLS_NIST.pem' + 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..") -- To view, visit https://gerrit.osmocom.org/c/pysim/+/40525?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: I6906732f7d193a9c2234075f4a82df5e0ed46100 Gerrit-Change-Number: 40525 Gerrit-PatchSet: 1 Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>

2 months, 2 weeks

1
0
0 0

[M] Change in pysim[master]: memory backed ephermal session store for easy concurrent runs

by Hoernchen

Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/40527?usp=email ) Change subject: memory backed ephermal session store for easy concurrent runs ...................................................................... memory backed ephermal session store for easy concurrent runs Change-Id: I05bfd6ff471ccf1c8c2b5f2b748b9d4125ddd4f7 --- M osmo-smdpp.py M pySim/esim/rsp.py 2 files changed, 63 insertions(+), 7 deletions(-) git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/27/40527/1 diff --git a/osmo-smdpp.py b/osmo-smdpp.py index 1bdc533..3fee0e0 100755 --- a/osmo-smdpp.py +++ b/osmo-smdpp.py @@ -409,7 +409,7 @@ continue return False - def __init__(self, server_hostname: str, ci_certs_path: str, common_cert_path: str, use_brainpool: bool = False): + def __init__(self, server_hostname: str, ci_certs_path: str, common_cert_path: str, use_brainpool: bool = False, in_memory: 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) @@ -430,9 +430,15 @@ 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')) - # 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" - self.rss = rsp.RspSessionStore(os.path.join(DATA_DIR, f"sm-dp-sessions-{session_db_suffix}")) + 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}") @app.handle_errors(ApiError) def handle_apierror(self, request: IRequest, failure): @@ -865,6 +871,8 @@ 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) args = parser.parse_args() logging.basicConfig(level=logging.DEBUG if args.verbose else logging.WARNING) diff --git a/pySim/esim/rsp.py b/pySim/esim/rsp.py index 3dccf2d..6ab3fa3 100644 --- a/pySim/esim/rsp.py +++ b/pySim/esim/rsp.py @@ -94,9 +94,57 @@ self.__dict__.update(state) -class RspSessionStore(shelve.DbfilenameShelf): - """A derived class as wrapper around the database-backed non-volatile storage 'shelve', in case we might - need to extend it in the future. We use it to store RspSessionState objects indexed by transactionId.""" +class RspSessionStore: + """A wrapper around the database-backed storage 'shelve' for storing RspSessionState objects. + Can be configured to use either file-based storage or in-memory storage. + We use it to store RspSessionState objects indexed by transactionId.""" + + 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 session store") + self._shelf = shelve.open(filename) + + # dunder magic + def __getitem__(self, key): + return self._shelf[key] + + def __setitem__(self, key, value): + 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 extract_euiccSigned1(authenticateServerResponse: bytes) -> bytes: """Extract the raw, DER-encoded binary euiccSigned1 field from the given AuthenticateServerResponse. This -- To view, visit https://gerrit.osmocom.org/c/pysim/+/40527?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: I05bfd6ff471ccf1c8c2b5f2b748b9d4125ddd4f7 Gerrit-Change-Number: 40527 Gerrit-PatchSet: 1 Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>

2 months, 2 weeks

1
0
0 0

[M] Change in pysim[master]: smdpp: fix asn1tool OBJECT IDENTIFIER decoding

by Hoernchen

Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/40526?usp=email ) Change subject: smdpp: fix asn1tool OBJECT IDENTIFIER decoding ...................................................................... smdpp: fix asn1tool OBJECT IDENTIFIER decoding Change-Id: Ic678e6c4a4c1a01de87a8dce26f4a5e452e8562a --- M osmo-smdpp.py 1 file changed, 87 insertions(+), 3 deletions(-) git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/26/40526/1 diff --git a/osmo-smdpp.py b/osmo-smdpp.py index 41c3562..1bdc533 100755 --- a/osmo-smdpp.py +++ b/osmo-smdpp.py @@ -17,6 +17,93 @@ # 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 @@ -37,7 +124,6 @@ from base64 import b64decode from klein import Klein from twisted.web.iweb import IRequest -import asn1tools from osmocom.utils import h2b, b2h, swap_nibbles @@ -140,8 +226,6 @@ if isinstance(ext_der, bytes): try: - import asn1tools - permitted_eins_schema = """ PermittedEins DEFINITIONS ::= BEGIN PermittedEins ::= SEQUENCE OF PrintableString -- 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: newchange Gerrit-Project: pysim Gerrit-Branch: master Gerrit-Change-Id: Ic678e6c4a4c1a01de87a8dce26f4a5e452e8562a Gerrit-Change-Number: 40526 Gerrit-PatchSet: 1 Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>

2 months, 2 weeks

1
0
0 0

[L] Change in pysim[master]: smdpp: proper checks, proper check order, test mode

by Hoernchen

Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/40528?usp=email ) Change subject: smdpp: proper checks, proper check order, test mode ...................................................................... smdpp: proper checks, proper check order, test mode Change-Id: Ie4db65594b2eaf6c95ffc5e73ff9ae61c4d9d3a3 --- M osmo-smdpp.py M pySim/esim/x509_cert.py A pySim/esim/x509_err.py 3 files changed, 839 insertions(+), 107 deletions(-) git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/28/40528/1 diff --git a/osmo-smdpp.py b/osmo-smdpp.py index 3fee0e0..edc6e85 100755 --- a/osmo-smdpp.py +++ b/osmo-smdpp.py @@ -110,6 +110,7 @@ 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 @@ -121,6 +122,7 @@ from pprint import pprint as pp import base64 +import time from base64 import b64decode from klein import Klein from twisted.web.iweb import IRequest @@ -131,7 +133,11 @@ 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 pySim.esim.x509_cert import CertAndPrivkey, CertificateSet, cert_get_subject_key_id, VerifyError +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__) @@ -171,18 +177,18 @@ for policy in cert_policies_ext.value: policy_oid = policy.policy_identifier.dotted_string - print(f"Found certificate policy: {policy_oid}") + logger.debug(f"Found certificate policy: {policy_oid}") if policy_oid == '2.23.146.1.2.1.2': - print("Detected EUM certificate variant: O (old)") + logger.debug("Detected EUM certificate variant: O (old)") return 'O' elif policy_oid == '2.23.146.1.2.1.0.0.0': - print("Detected EUM certificate variant: Ov3/A/B/C (new)") + logger.debug("Detected EUM certificate variant: Ov3/A/B/C (new)") return 'NEW' except x509.ExtensionNotFound: - print("No Certificate Policies extension found") + logger.debug("No Certificate Policies extension found") except Exception as e: - print(f"Error checking certificate policies: {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 @@ -205,7 +211,7 @@ unique_iins = list(set(permitted_iins)) - print(f"Total unique permitted IINs found: {len(unique_iins)}") + logger.debug(f"Total unique permitted IINs found: {len(unique_iins)}") return unique_iins def _parse_gsma_permitted_eins(eum_cert) -> List[str]: @@ -219,7 +225,7 @@ for ext in eum_cert.extensions: if ext.oid == permitted_eins_oid: - print(f"Found GSMA permittedEins extension: {ext.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 @@ -243,14 +249,14 @@ all(c in '0123456789ABCDEF' for c in iin_clean) and len(iin_clean) % 2 == 0): permitted_iins.append(iin_clean) - print(f"Found permitted IIN (GSMA): {iin_clean}") + logger.debug(f"Found permitted IIN (GSMA): {iin_clean}") else: - print(f"Invalid IIN format: {iin_string} (cleaned: {iin_clean})") + logger.debug(f"Invalid IIN format: {iin_string} (cleaned: {iin_clean})") except Exception as e: - print(f"Error parsing GSMA permittedEins extension: {e}") + logger.debug(f"Error parsing GSMA permittedEins extension: {e}") except Exception as e: - print(f"Error accessing GSMA certificate extensions: {e}") + logger.debug(f"Error accessing GSMA certificate extensions: {e}") return permitted_iins @@ -283,12 +289,12 @@ all(c in '0123456789ABCDEF' for c in serial_value) and len(serial_value) % 2 == 0): permitted_iins.append(serial_value) - print(f"Found permitted IIN (nameConstraints/DN): {serial_value}") + logger.debug(f"Found permitted IIN (nameConstraints/DN): {serial_value}") except x509.ExtensionNotFound: - print("No nameConstraints extension found") + logger.debug("No nameConstraints extension found") except Exception as e: - print(f"Error parsing nameConstraints: {e}") + logger.debug(f"Error parsing nameConstraints: {e}") return permitted_iins @@ -296,29 +302,29 @@ 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: - print(f"Invalid EID format: {eid}") + logger.debug(f"Invalid EID format: {eid}") return False try: permitted_eins = parse_permitted_eins_from_cert(eum_cert) if not permitted_eins: - print("Warning: No permitted EINs found in EUM certificate") + logger.debug("Warning: No permitted EINs found in EUM certificate") return False eid_normalized = eid.upper() - print(f"Validating EID {eid_normalized} against {len(permitted_eins)} permitted EINs") + 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): - print(f"EID {eid_normalized} matches permitted EIN {permitted_ein}") + logger.debug(f"EID {eid_normalized} matches permitted EIN {permitted_ein}") return True - print(f"EID {eid_normalized} is not in any permitted EIN list") + logger.debug(f"EID {eid_normalized} is not in any permitted EIN list") return False except Exception as e: - print(f"Error validating EID: {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: @@ -348,6 +354,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): @@ -405,14 +546,20 @@ 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 {} + + 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 @@ -439,6 +586,231 @@ 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): @@ -479,21 +851,41 @@ @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: 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) - #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 ... @@ -501,25 +893,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: + # 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() @@ -535,9 +938,9 @@ 'serverAddress': self.server_hostname, 'serverChallenge': serverChallenge, } - print("Tx serverSigned1: %s" % serverSigned1) + logger.debug("Tx serverSigned1: %s" % serverSigned1) serverSigned1_bin = rsp.asn1.encode('ServerSigned1', serverSigned1) - print("Tx serverSigned1: %s" % rsp.asn1.decode('ServerSigned1', serverSigned1_bin)) + logger.debug("Tx serverSigned1: %s" % rsp.asn1.decode('ServerSigned1', serverSigned1_bin)) output = {} output['serverSigned1'] = b64encode2str(serverSigned1_bin) @@ -588,69 +991,264 @@ 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: + # 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. - # 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 - print("EID (from eUICC cert): %s" % ss.eid) + 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: + # 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') @@ -658,19 +1256,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) @@ -693,6 +1297,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']) @@ -703,7 +1308,10 @@ 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 + 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] @@ -711,11 +1319,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'] @@ -723,23 +1333,59 @@ # 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 - print("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()))) + 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) - print("shared_secret: %s" % b2h(ss.shared_secret)) + 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 @@ -776,15 +1422,18 @@ transactionId = b2h(pird['transactionId']) ss = self.rss.get(transactionId, None) if ss is None: - print("Unable to find session for transactionId") - return + # 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): - raise Exception('ECDSA signature verification failed on notification') - print("Profile Installation Final Result: ", pird['finalResult']) + # 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': @@ -802,16 +1451,20 @@ 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 iccid = other_notif.get('iccid', None) if iccid: iccid = swap_nibbles(b2h(iccid)) - print("handleNotification: EID %s: %s of %s" % (eid, pmo, 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 @@ -828,7 +1481,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) @@ -843,15 +1497,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. @@ -859,7 +1532,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): @@ -873,12 +1547,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..f2f66fe 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() 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/+/40528?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: Ie4db65594b2eaf6c95ffc5e73ff9ae61c4d9d3a3 Gerrit-Change-Number: 40528 Gerrit-PatchSet: 1 Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>

2 months, 2 weeks

1
0
0 0

[S] Change in pysim[master]: smdpp: reorder imports

by Hoernchen

Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/40523?usp=email ) Change subject: smdpp: reorder imports ...................................................................... smdpp: reorder imports Change-Id: Ib72089fb75d71f0d33c9ea17e5491dd52558f532 --- M osmo-smdpp.py 1 file changed, 7 insertions(+), 6 deletions(-) git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/23/40523/1 diff --git a/osmo-smdpp.py b/osmo-smdpp.py index b669ff2..82fb201 100755 --- a/osmo-smdpp.py +++ b/osmo-smdpp.py @@ -17,6 +17,13 @@ # 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/>. +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 pathlib import Path import json import sys import argparse @@ -245,12 +252,6 @@ if status_code_data: js['header']['functionExecutionStatus']['statusCodeData'] = status_code_data -from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature, encode_dss_signature -from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat, PrivateFormat, NoEncryption -from cryptography.hazmat.primitives.asymmetric import ec -from cryptography.hazmat.primitives import hashes -from cryptography.exceptions import InvalidSignature -from cryptography import x509 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.""" -- To view, visit https://gerrit.osmocom.org/c/pysim/+/40523?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: Ib72089fb75d71f0d33c9ea17e5491dd52558f532 Gerrit-Change-Number: 40523 Gerrit-PatchSet: 1 Gerrit-Owner: Hoernchen <ewild(a)sysmocom.de>

2 months, 2 weeks

1
0
0 0

2025

2024

2023

2022

gerrit-log June 2025