Attention is currently required from: laforge.
lynxis lazus has posted comments on this change. ( https://gerrit.osmocom.org/c/pysim/+/36960?usp=email )
Change subject: esim.saip: Implement ProfileElement.header_name for more PE types
......................................................................
Patch Set 3: Code-Review+1
(1 comment)
File pySim/esim/saip/__init__.py:
https://gerrit.osmocom.org/c/pysim/+/36960/comment/28722650_d5a1ea02
PS3, Line 176: elif self.type == 'genericFileManagement':
you might use a dict for this.
elif self.type in some_dict:
return some_dict[self.type]
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/36960?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I37951a0441fe53fce7a329066aebd973389cb743
Gerrit-Change-Number: 36960
Gerrit-PatchSet: 3
Gerrit-Owner: laforge <laforge(a)osmocom.org>
Gerrit-Reviewer: Jenkins Builder
Gerrit-Reviewer: lynxis lazus <lynxis(a)fe80.eu>
Gerrit-Attention: laforge <laforge(a)osmocom.org>
Gerrit-Comment-Date: Sun, 02 Jun 2024 15:03:33 +0000
Gerrit-HasComments: Yes
Gerrit-Has-Labels: Yes
Gerrit-MessageType: comment
Attention is currently required from: fixeria.
lynxis lazus has posted comments on this change. ( https://gerrit.osmocom.org/c/erlang/osmo-epdg/+/36968?usp=email )
Change subject: README.md: fix copy-pasted 'osmo-bsc' and a broken link
......................................................................
Patch Set 1: Code-Review+2
--
To view, visit https://gerrit.osmocom.org/c/erlang/osmo-epdg/+/36968?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings
Gerrit-Project: erlang/osmo-epdg
Gerrit-Branch: master
Gerrit-Change-Id: I760db277ee682298d8c7a4d11cf68c86d94fe368
Gerrit-Change-Number: 36968
Gerrit-PatchSet: 1
Gerrit-Owner: fixeria <vyanitskiy(a)sysmocom.de>
Gerrit-Reviewer: Jenkins Builder
Gerrit-Reviewer: lynxis lazus <lynxis(a)fe80.eu>
Gerrit-Attention: fixeria <vyanitskiy(a)sysmocom.de>
Gerrit-Comment-Date: Sun, 02 Jun 2024 14:57:49 +0000
Gerrit-HasComments: No
Gerrit-Has-Labels: Yes
Gerrit-MessageType: comment
laforge has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/36969?usp=email )
Change subject: utils: Introduce BER-TLV parsers that return raw tag or even raw TLV
......................................................................
utils: Introduce BER-TLV parsers that return raw tag or even raw TLV
In the eSIM RSP univers there are some rather ugly layering violatoins
where ASN.1 cannot be parsed but we have to mess with raw TLVs and the
details of DER encoding. Let's add two funtions that make it more
convenient to work with this: They return the raw tag as integer, or
even the entire encoded TLV rather than the value part only.
Change-Id: I1e68a4003b833e86e9282c77325afa86ce144b98
---
M pySim/esim/rsp.py
M pySim/utils.py
2 files changed, 56 insertions(+), 21 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/69/36969/1
diff --git a/pySim/esim/rsp.py b/pySim/esim/rsp.py
index a032031..4230d1f 100644
--- a/pySim/esim/rsp.py
+++ b/pySim/esim/rsp.py
@@ -24,7 +24,7 @@
from cryptography.hazmat.primitives.serialization import Encoding
from cryptography import x509
-from pySim.utils import bertlv_parse_one, bertlv_encode_tag, bertlv_encode_len, b2h
+from pySim.utils import bertlv_parse_one_rawtag, bertlv_return_one_rawtlv, b2h
from pySim.esim import compile_asn1_subdir
asn1 = compile_asn1_subdir('rsp')
@@ -101,37 +101,31 @@
def extract_euiccSigned1(authenticateServerResponse: bytes) -> bytes:
"""Extract the raw, DER-encoded binary euiccSigned1 field from the given AuthenticateServerResponse. This
is needed due to the very peculiar SGP.22 notion of signing sections of DER-encoded ASN.1 objects."""
- tdict, l, v, remainder = bertlv_parse_one(authenticateServerResponse)
- rawtag = bertlv_encode_tag(tdict)
+ rawtag, l, v, remainder = bertlv_parse_one_rawtag(authenticateServerResponse)
if len(remainder):
raise ValueError('Excess data at end of TLV')
- if b2h(rawtag) != 'bf38':
+ if rawtag != 0xbf38:
raise ValueError('Unexpected outer tag: %s' % b2h(rawtag))
- tdict, l, v1, remainder = bertlv_parse_one(v)
- rawtag = bertlv_encode_tag(tdict)
- if b2h(rawtag) != 'a0':
+ rawtag, l, v1, remainder = bertlv_parse_one_rawtag(v)
+ if rawtag != 0xa0:
raise ValueError('Unexpected tag where CHOICE was expected')
- tdict, l, v2, remainder = bertlv_parse_one(v1)
- rawtag = bertlv_encode_tag(tdict)
- if b2h(rawtag) != '30':
+ rawtag, l, tlv2, remainder = bertlv_return_one_rawtlv(v1)
+ if rawtag != 0x30:
raise ValueError('Unexpected tag where SEQUENCE was expected')
- return rawtag + bertlv_encode_len(l) + v2
+ return tlv2
def extract_euiccSigned2(prepareDownloadResponse: bytes) -> bytes:
"""Extract the raw, DER-encoded binary euiccSigned2 field from the given prepareDownloadrResponse. This is
needed due to the very peculiar SGP.22 notion of signing sections of DER-encoded ASN.1 objects."""
- tdict, l, v, remainder = bertlv_parse_one(prepareDownloadResponse)
- rawtag = bertlv_encode_tag(tdict)
+ rawtag, l, v, remainder = bertlv_parse_one_rawtag(prepareDownloadResponse)
if len(remainder):
raise ValueError('Excess data at end of TLV')
- if b2h(rawtag) != 'bf21':
+ if rawtag != 0xbf21:
raise ValueError('Unexpected outer tag: %s' % b2h(rawtag))
- tdict, l, v1, remainder = bertlv_parse_one(v)
- rawtag = bertlv_encode_tag(tdict)
- if b2h(rawtag) != 'a0':
+ rawtag, l, v1, remainder = bertlv_parse_one_rawtag(v)
+ if rawtag != 0xa0:
raise ValueError('Unexpected tag where CHOICE was expected')
- tdict, l, v2, remainder = bertlv_parse_one(v1)
- rawtag = bertlv_encode_tag(tdict)
- if b2h(rawtag) != '30':
+ rawtag, l, tlv2, remainder = bertlv_return_one_rawtlv(v1)
+ if rawtag != 0x30:
raise ValueError('Unexpected tag where SEQUENCE was expected')
- return rawtag + bertlv_encode_len(l) + v2
+ return tlv2
diff --git a/pySim/utils.py b/pySim/utils.py
index afa476b..2362b59 100644
--- a/pySim/utils.py
+++ b/pySim/utils.py
@@ -359,6 +359,32 @@
remainder = remainder[length:]
return (tagdict, length, value, remainder)
+def bertlv_parse_one_rawtag(binary: bytes) -> Tuple[int, int, bytes, bytes]:
+ """Parse a single TLV IE at the start of the given binary data; return tag as raw integer.
+ Args:
+ binary : binary input data of BER-TLV length field
+ Returns:
+ Tuple of (tag:int, len:int, remainder:bytes)
+ """
+ (tag, remainder) = bertlv_parse_tag_raw(binary)
+ (length, remainder) = bertlv_parse_len(remainder)
+ value = remainder[:length]
+ remainder = remainder[length:]
+ return (tag, length, value, remainder)
+
+def bertlv_return_one_rawtlv(binary: bytes) -> Tuple[int, int, bytes, bytes]:
+ """Return one single [encoded] TLV IE at the start of the given binary data.
+ Args:
+ binary : binary input data of BER-TLV length field
+ Returns:
+ Tuple of (tag:int, len:int, tlv:bytes, remainder:bytes)
+ """
+ (tag, remainder) = bertlv_parse_tag_raw(binary)
+ (length, remainder) = bertlv_parse_len(remainder)
+ tl_length = len(binary) - len(remainder)
+ value = binary[:tl_length] + remainder[:length]
+ remainder = remainder[length:]
+ return (tag, length, value, remainder)
def dgi_parse_tag_raw(binary: bytes) -> Tuple[int, bytes]:
# In absence of any clear spec guidance we assume it's always 16 bit
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/36969?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I1e68a4003b833e86e9282c77325afa86ce144b98
Gerrit-Change-Number: 36969
Gerrit-PatchSet: 1
Gerrit-Owner: laforge <laforge(a)osmocom.org>
Gerrit-MessageType: newchange
laforge has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/36971?usp=email )
Change subject: [cosmetic] fix typos in comments
......................................................................
[cosmetic] fix typos in comments
Change-Id: I549ef7002e6ebef3f13af620cad8d03c7f4d891a
---
M osmo-smdpp.py
M pySim/esim/rsp.py
2 files changed, 11 insertions(+), 2 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/71/36971/1
diff --git a/osmo-smdpp.py b/osmo-smdpp.py
index d7fc872..9551396 100755
--- a/osmo-smdpp.py
+++ b/osmo-smdpp.py
@@ -444,7 +444,7 @@
ss.host_id = b'mahlzeit'
- # Generate Session Keys using the CRT, opPK.eUICC.ECKA and otSK.DP.ECKA according to annex G
+ # 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))
diff --git a/pySim/esim/rsp.py b/pySim/esim/rsp.py
index 4230d1f..c2a163b 100644
--- a/pySim/esim/rsp.py
+++ b/pySim/esim/rsp.py
@@ -37,7 +37,7 @@
def __init__(self, transactionId: str, serverChallenge: bytes, ci_cert_id: bytes):
self.transactionId = transactionId
self.serverChallenge = serverChallenge
- # used at a later point between API calsl
+ # used at a later point between API calls
self.ci_cert_id = ci_cert_id
self.euicc_cert: Optional[x509.Certificate] = None
self.eum_cert: Optional[x509.Certificate] = None
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/36971?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I549ef7002e6ebef3f13af620cad8d03c7f4d891a
Gerrit-Change-Number: 36971
Gerrit-PatchSet: 1
Gerrit-Owner: laforge <laforge(a)osmocom.org>
Gerrit-MessageType: newchange
laforge has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/36970?usp=email )
Change subject: esim.bsp: Fix a bug in demac_only_one()
......................................................................
esim.bsp: Fix a bug in demac_only_one()
When de-MAC-ing at the recipient side, we must increment the cipher(!)
block number even if no ciphering is done at all.
We did this correctly for MAC (sender) case, but not on the de-MAC
(receiver) case.
Change-Id: I97993f9e8357b36401d435aaa15558d1c7e411eb
---
M pySim/esim/bsp.py
1 file changed, 17 insertions(+), 0 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/70/36970/1
diff --git a/pySim/esim/bsp.py b/pySim/esim/bsp.py
index 2afbd46..81fe092 100644
--- a/pySim/esim/bsp.py
+++ b/pySim/esim/bsp.py
@@ -287,6 +287,8 @@
def demac_only_one(self, ciphertext: bytes) -> bytes:
payload = self.m_algo.verify(ciphertext)
_tdict, _l, val, _remain = bertlv_parse_one(payload)
+ # The data block counter for ICV caluclation is incremented also for each segment with C-MAC only.
+ self.c_algo.block_nr += 1
return val
def demac_only(self, ciphertext_list: List[bytes]) -> bytes:
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/36970?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I97993f9e8357b36401d435aaa15558d1c7e411eb
Gerrit-Change-Number: 36970
Gerrit-PatchSet: 1
Gerrit-Owner: laforge <laforge(a)osmocom.org>
Gerrit-MessageType: newchange
laforge has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/36972?usp=email )
Change subject: esim.es2p: Split generic part of HTTP/REST API from ES2+
......................................................................
esim.es2p: Split generic part of HTTP/REST API from ES2+
This way we can reuse it for other eSIM RSP HTTP interfaces like
ES9+, ES11, ...
Change-Id: I468041da40a88875e8df15b04d3ad508e06f16f7
---
M pySim/esim/es2p.py
A pySim/esim/http_json_api.py
2 files changed, 276 insertions(+), 230 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/72/36972/1
diff --git a/pySim/esim/es2p.py b/pySim/esim/es2p.py
index fa21d2c..b026f9e 100644
--- a/pySim/esim/es2p.py
+++ b/pySim/esim/es2p.py
@@ -23,90 +23,11 @@
import time
import base64
+from pySim.esim.http_json_api import *
+
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
-class ApiParam(abc.ABC):
- """A class reprsenting a single parameter in the ES2+ API."""
- @classmethod
- def verify_decoded(cls, data):
- """Verify the decoded reprsentation of a value. Should raise an exception if somthing is odd."""
- pass
-
- @classmethod
- def verify_encoded(cls, data):
- """Verify the encoded reprsentation of a value. Should raise an exception if somthing is odd."""
- pass
-
- @classmethod
- def encode(cls, data):
- """[Validate and] Encode the given value."""
- cls.verify_decoded(data)
- encoded = cls._encode(data)
- cls.verify_decoded(encoded)
- return encoded
-
- @classmethod
- def _encode(cls, data):
- """encoder function, typically [but not always] overridden by derived class."""
- return data
-
- @classmethod
- def decode(cls, data):
- """[Validate and] Decode the given value."""
- cls.verify_encoded(data)
- decoded = cls._decode(data)
- cls.verify_decoded(decoded)
- return decoded
-
- @classmethod
- def _decode(cls, data):
- """decoder function, typically [but not always] overridden by derived class."""
- return data
-
-class ApiParamString(ApiParam):
- """Base class representing an API parameter of 'string' type."""
- pass
-
-
-class ApiParamInteger(ApiParam):
- """Base class representing an API parameter of 'integer' type."""
- @classmethod
- def _decode(cls, data):
- return int(data)
-
- @classmethod
- def _encode(cls, data):
- return str(data)
-
- @classmethod
- def verify_decoded(cls, data):
- if not isinstance(data, int):
- raise TypeError('Expected an integer input data type')
-
- @classmethod
- def verify_encoded(cls, data):
- if isinstance(data, int):
- return
- if not data.isdecimal():
- raise ValueError('integer (%s) contains non-decimal characters' % data)
- assert str(int(data)) == data
-
-class ApiParamBoolean(ApiParam):
- """Base class representing an API parameter of 'boolean' type."""
- @classmethod
- def _encode(cls, data):
- return bool(data)
-
-class ApiParamFqdn(ApiParam):
- """String, as a list of domain labels concatenated using the full stop (dot, period) character as
- separator between labels. Labels are restricted to the Alphanumeric mode character set defined in table 5
- of ISO/IEC 18004"""
- @classmethod
- def verify_encoded(cls, data):
- # FIXME
- pass
-
class param:
class Iccid(ApiParamString):
"""String representation of 19 or 20 digits, where the 20th digit MAY optionally be the padding
@@ -172,9 +93,6 @@
class SmdsAddress(ApiParamFqdn):
pass
- class SmdpAddress(ApiParamFqdn):
- pass
-
class ReleaseFlag(ApiParamBoolean):
pass
@@ -197,150 +115,12 @@
class NotificationPointStatus(ApiParam):
pass
- class ResultData(ApiParam):
- @classmethod
- def _decode(cls, data):
- return base64.b64decode(data)
+ class ResultData(ApiParamBase64):
+ pass
- @classmethod
- def _encode(cls, data):
- return base64.b64encode(data)
-
- class JsonResponseHeader(ApiParam):
- """SGP.22 section 6.5.1.4."""
- @classmethod
- def verify_decoded(cls, data):
- fe_status = data.get('functionExecutionStatus')
- if not fe_status:
- raise ValueError('Missing mandatory functionExecutionStatus in header')
- status = fe_status.get('status')
- if not status:
- raise ValueError('Missing mandatory status in header functionExecutionStatus')
- if status not in ['Executed-Success', 'Executed-WithWarning', 'Failed', 'Expired']:
- raise ValueError('Unknown/unspecified status "%s"' % status)
-
-
-class HttpStatusError(Exception):
- pass
-
-class HttpHeaderError(Exception):
- pass
-
-class Es2PlusApiError(Exception):
- """Exception representing an error at the ES2+ API level (status != Executed)."""
- def __init__(self, func_ex_status: dict):
- self.status = func_ex_status['status']
- sec = {
- 'subjectCode': None,
- 'reasonCode': None,
- 'subjectIdentifier': None,
- 'message': None,
- }
- actual_sec = func_ex_status.get('statusCodeData', None)
- sec.update(actual_sec)
- self.subject_code = sec['subjectCode']
- self.reason_code = sec['reasonCode']
- self.subject_id = sec['subjectIdentifier']
- self.message = sec['message']
-
- def __str__(self):
- return f'{self.status}("{self.subject_code}","{self.reason_code}","{self.subject_id}","{self.message}")'
-
-class Es2PlusApiFunction(abc.ABC):
+class Es2PlusApiFunction(JsonHttpApiFunction):
"""Base classs for representing an ES2+ API Function."""
- # the below class variables are expected to be overridden in derived classes
-
- path = None
- # dictionary of input parameters. key is parameter name, value is ApiParam class
- input_params = {}
- # list of mandatory input parameters
- input_mandatory = []
- # dictionary of output parameters. key is parameter name, value is ApiParam class
- output_params = {}
- # list of mandatory output parameters (for successful response)
- output_mandatory = []
- # expected HTTP status code of the response
- expected_http_status = 200
- # the HTTP method used (GET, OPTIONS, HEAD, POST, PUT, PATCH or DELETE)
- http_method = 'POST'
-
- def __init__(self, url_prefix: str, func_req_id: str, session):
- self.url_prefix = url_prefix
- self.func_req_id = func_req_id
- self.session = session
-
- def encode(self, data: dict, func_call_id: str) -> dict:
- """Validate an encode input dict into JSON-serializable dict for request body."""
- output = {
- 'header': {
- 'functionRequesterIdentifier': self.func_req_id,
- 'functionCallIdentifier': func_call_id
- }
- }
- for p in self.input_mandatory:
- if not p in data:
- raise ValueError('Mandatory input parameter %s missing' % p)
- for p, v in data.items():
- p_class = self.input_params.get(p)
- if not p_class:
- logger.warning('Unexpected/unsupported input parameter %s=%s', p, v)
- output[p] = v
- else:
- output[p] = p_class.encode(v)
- return output
-
-
- def decode(self, data: dict) -> dict:
- """[further] Decode and validate the JSON-Dict of the respnse body."""
- output = {}
- # let's first do the header, it's special
- if not 'header' in data:
- raise ValueError('Mandatory output parameter "header" missing')
- hdr_class = self.output_params.get('header')
- output['header'] = hdr_class.decode(data['header'])
-
- if output['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
- raise Es2PlusApiError(output['header']['functionExecutionStatus'])
- # we can only expect mandatory parameters to be present in case of successful execution
- for p in self.output_mandatory:
- if p == 'header':
- continue
- if not p in data:
- raise ValueError('Mandatory output parameter "%s" missing' % p)
- for p, v in data.items():
- p_class = self.output_params.get(p)
- if not p_class:
- logger.warning('Unexpected/unsupported output parameter "%s"="%s"', p, v)
- output[p] = v
- else:
- output[p] = p_class.decode(v)
- return output
-
- def call(self, data: dict, func_call_id:str, timeout=10) -> dict:
- """Make an API call to the ES2+ API endpoint represented by this object.
- Input data is passed in `data` as json-serializable dict. Output data
- is returned as json-deserialized dict."""
- url = self.url_prefix + self.path
- encoded = json.dumps(self.encode(data, func_call_id))
- headers = {
- 'Content-Type': 'application/json',
- 'X-Admin-Protocol': 'gsma/rsp/v2.5.0',
- }
-
- logger.debug("HTTP REQ %s - '%s'" % (url, encoded))
- response = self.session.request(self.http_method, url, data=encoded, headers=headers, timeout=timeout)
- logger.debug("HTTP RSP-STS: [%u] hdr: %s" % (response.status_code, response.headers))
- logger.debug("HTTP RSP: %s" % (response.content))
-
- if response.status_code != self.expected_http_status:
- raise HttpStatusError(response)
- if not response.headers.get('Content-Type').startswith(headers['Content-Type']):
- raise HttpHeaderError(response)
- if not response.headers.get('X-Admin-Protocol', 'gsma/rsp/v2.unknown').startswith('gsma/rsp/v2.'):
- raise HttpHeaderError(response)
-
- return self.decode(response.json())
-
+ pass
# ES2+ DownloadOrder function (SGP.22 section 5.3.1)
class DownloadOrder(Es2PlusApiFunction):
@@ -351,7 +131,7 @@
'profileType': param.ProfileType
}
output_params = {
- 'header': param.JsonResponseHeader,
+ 'header': JsonResponseHeader,
'iccid': param.Iccid,
}
output_mandatory = ['header', 'iccid']
@@ -369,7 +149,7 @@
}
input_mandatory = ['iccid', 'releaseFlag']
output_params = {
- 'header': param.JsonResponseHeader,
+ 'header': JsonResponseHeader,
'eid': param.Eid,
'matchingId': param.MatchingId,
'smdpAddress': param.SmdpAddress,
@@ -387,7 +167,7 @@
}
input_mandatory = ['finalProfileStatusIndicator', 'iccid']
output_params = {
- 'header': param.JsonResponseHeader,
+ 'header': JsonResponseHeader,
}
output_mandatory = ['header']
@@ -399,7 +179,7 @@
}
input_mandatory = ['iccid']
output_params = {
- 'header': param.JsonResponseHeader,
+ 'header': JsonResponseHeader,
}
output_mandatory = ['header']
diff --git a/pySim/esim/http_json_api.py b/pySim/esim/http_json_api.py
new file mode 100644
index 0000000..ec9cd00
--- /dev/null
+++ b/pySim/esim/http_json_api.py
@@ -0,0 +1,254 @@
+"""GSMA eSIM RSP HTTP/REST/JSON interface according to SGP.22 v2.5"""
+
+# (C) 2024 by Harald Welte <laforge(a)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/>.
+
+import abc
+import requests
+import logging
+import json
+from datetime import datetime
+import time
+import base64
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+class ApiParam(abc.ABC):
+ """A class reprsenting a single parameter in the API."""
+ @classmethod
+ def verify_decoded(cls, data):
+ """Verify the decoded reprsentation of a value. Should raise an exception if somthing is odd."""
+ pass
+
+ @classmethod
+ def verify_encoded(cls, data):
+ """Verify the encoded reprsentation of a value. Should raise an exception if somthing is odd."""
+ pass
+
+ @classmethod
+ def encode(cls, data):
+ """[Validate and] Encode the given value."""
+ cls.verify_decoded(data)
+ encoded = cls._encode(data)
+ cls.verify_decoded(encoded)
+ return encoded
+
+ @classmethod
+ def _encode(cls, data):
+ """encoder function, typically [but not always] overridden by derived class."""
+ return data
+
+ @classmethod
+ def decode(cls, data):
+ """[Validate and] Decode the given value."""
+ cls.verify_encoded(data)
+ decoded = cls._decode(data)
+ cls.verify_decoded(decoded)
+ return decoded
+
+ @classmethod
+ def _decode(cls, data):
+ """decoder function, typically [but not always] overridden by derived class."""
+ return data
+
+class ApiParamString(ApiParam):
+ """Base class representing an API parameter of 'string' type."""
+ pass
+
+
+class ApiParamInteger(ApiParam):
+ """Base class representing an API parameter of 'integer' type."""
+ @classmethod
+ def _decode(cls, data):
+ return int(data)
+
+ @classmethod
+ def _encode(cls, data):
+ return str(data)
+
+ @classmethod
+ def verify_decoded(cls, data):
+ if not isinstance(data, int):
+ raise TypeError('Expected an integer input data type')
+
+ @classmethod
+ def verify_encoded(cls, data):
+ if isinstance(data, int):
+ return
+ if not data.isdecimal():
+ raise ValueError('integer (%s) contains non-decimal characters' % data)
+ assert str(int(data)) == data
+
+class ApiParamBoolean(ApiParam):
+ """Base class representing an API parameter of 'boolean' type."""
+ @classmethod
+ def _encode(cls, data):
+ return bool(data)
+
+class ApiParamFqdn(ApiParam):
+ """String, as a list of domain labels concatenated using the full stop (dot, period) character as
+ separator between labels. Labels are restricted to the Alphanumeric mode character set defined in table 5
+ of ISO/IEC 18004"""
+ @classmethod
+ def verify_encoded(cls, data):
+ # FIXME
+ pass
+
+class ApiParamBase64(ApiParam):
+ @classmethod
+ def _decode(cls, data):
+ return base64.b64decode(data)
+
+ @classmethod
+ def _encode(cls, data):
+ return base64.b64encode(data).decode('ascii')
+
+class SmdpAddress(ApiParamFqdn):
+ pass
+
+class JsonResponseHeader(ApiParam):
+ """SGP.22 section 6.5.1.4."""
+ @classmethod
+ def verify_decoded(cls, data):
+ fe_status = data.get('functionExecutionStatus')
+ if not fe_status:
+ raise ValueError('Missing mandatory functionExecutionStatus in header')
+ status = fe_status.get('status')
+ if not status:
+ raise ValueError('Missing mandatory status in header functionExecutionStatus')
+ if status not in ['Executed-Success', 'Executed-WithWarning', 'Failed', 'Expired']:
+ raise ValueError('Unknown/unspecified status "%s"' % status)
+
+
+class HttpStatusError(Exception):
+ pass
+
+class HttpHeaderError(Exception):
+ pass
+
+class ApiError(Exception):
+ """Exception representing an error at the API level (status != Executed)."""
+ def __init__(self, func_ex_status: dict):
+ self.status = func_ex_status['status']
+ sec = {
+ 'subjectCode': None,
+ 'reasonCode': None,
+ 'subjectIdentifier': None,
+ 'message': None,
+ }
+ actual_sec = func_ex_status.get('statusCodeData', None)
+ sec.update(actual_sec)
+ self.subject_code = sec['subjectCode']
+ self.reason_code = sec['reasonCode']
+ self.subject_id = sec['subjectIdentifier']
+ self.message = sec['message']
+
+ def __str__(self):
+ return f'{self.status}("{self.subject_code}","{self.reason_code}","{self.subject_id}","{self.message}")'
+
+class JsonHttpApiFunction(abc.ABC):
+ """Base classs for representing an HTTP[s] API Function."""
+ # the below class variables are expected to be overridden in derived classes
+
+ path = None
+ # dictionary of input parameters. key is parameter name, value is ApiParam class
+ input_params = {}
+ # list of mandatory input parameters
+ input_mandatory = []
+ # dictionary of output parameters. key is parameter name, value is ApiParam class
+ output_params = {}
+ # list of mandatory output parameters (for successful response)
+ output_mandatory = []
+ # expected HTTP status code of the response
+ expected_http_status = 200
+ # the HTTP method used (GET, OPTIONS, HEAD, POST, PUT, PATCH or DELETE)
+ http_method = 'POST'
+
+ def __init__(self, url_prefix: str, func_req_id: str, session: requests.Session):
+ self.url_prefix = url_prefix
+ self.func_req_id = func_req_id
+ self.session = session
+
+ def encode(self, data: dict, func_call_id: str) -> dict:
+ """Validate an encode input dict into JSON-serializable dict for request body."""
+ output = {
+ 'header': {
+ 'functionRequesterIdentifier': self.func_req_id,
+ 'functionCallIdentifier': func_call_id
+ }
+ }
+ for p in self.input_mandatory:
+ if not p in data:
+ raise ValueError('Mandatory input parameter %s missing' % p)
+ for p, v in data.items():
+ p_class = self.input_params.get(p)
+ if not p_class:
+ logger.warning('Unexpected/unsupported input parameter %s=%s', p, v)
+ output[p] = v
+ else:
+ output[p] = p_class.encode(v)
+ return output
+
+ def decode(self, data: dict) -> dict:
+ """[further] Decode and validate the JSON-Dict of the respnse body."""
+ output = {}
+ # let's first do the header, it's special
+ if not 'header' in data:
+ raise ValueError('Mandatory output parameter "header" missing')
+ hdr_class = self.output_params.get('header')
+ output['header'] = hdr_class.decode(data['header'])
+
+ if output['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
+ raise ApiError(output['header']['functionExecutionStatus'])
+ # we can only expect mandatory parameters to be present in case of successful execution
+ for p in self.output_mandatory:
+ if p == 'header':
+ continue
+ if not p in data:
+ raise ValueError('Mandatory output parameter "%s" missing' % p)
+ for p, v in data.items():
+ p_class = self.output_params.get(p)
+ if not p_class:
+ logger.warning('Unexpected/unsupported output parameter "%s"="%s"', p, v)
+ output[p] = v
+ else:
+ output[p] = p_class.decode(v)
+ return output
+
+ def call(self, data: dict, func_call_id:str, timeout=10) -> dict:
+ """Make an API call to the HTTP API endpoint represented by this object.
+ Input data is passed in `data` as json-serializable dict. Output data
+ is returned as json-deserialized dict."""
+ url = self.url_prefix + self.path
+ encoded = json.dumps(self.encode(data, func_call_id))
+ headers = {
+ 'Content-Type': 'application/json',
+ 'X-Admin-Protocol': 'gsma/rsp/v2.5.0',
+ }
+
+ logger.debug("HTTP REQ %s - '%s'" % (url, encoded))
+ response = self.session.request(self.http_method, url, data=encoded, headers=headers, timeout=timeout)
+ logger.debug("HTTP RSP-STS: [%u] hdr: %s" % (response.status_code, response.headers))
+ logger.debug("HTTP RSP: %s" % (response.content))
+
+ if response.status_code != self.expected_http_status:
+ raise HttpStatusError(response)
+ if not response.headers.get('Content-Type').startswith(headers['Content-Type']):
+ raise HttpHeaderError(response)
+ if not response.headers.get('X-Admin-Protocol', 'gsma/rsp/v2.unknown').startswith('gsma/rsp/v2.'):
+ raise HttpHeaderError(response)
+
+ return self.decode(response.json())
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/36972?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I468041da40a88875e8df15b04d3ad508e06f16f7
Gerrit-Change-Number: 36972
Gerrit-PatchSet: 1
Gerrit-Owner: laforge <laforge(a)osmocom.org>
Gerrit-MessageType: newchange
Attention is currently required from: laforge.
Hello Jenkins Builder,
I'd like you to reexamine a change. Please visit
https://gerrit.osmocom.org/c/pysim/+/36961?usp=email
to look at the new patch set (#3).
The following approvals got outdated and were removed:
Verified+1 by Jenkins Builder
Change subject: pySim.esim.saip: Meaningful constructors for [I]SD + SSD
......................................................................
pySim.esim.saip: Meaningful constructors for [I]SD + SSD
So far the main use case was to read a ProfileElement-SD from
a DER file. But when we want to construct one from scratch,
we need to have the constructor put some meaningful [default]
values into the class members.
Change-Id: I69e104f1d78165c12291317326dbab05977a1574
---
M pySim/esim/saip/__init__.py
1 file changed, 85 insertions(+), 1 deletion(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/61/36961/3
--
To view, visit https://gerrit.osmocom.org/c/pysim/+/36961?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings
Gerrit-Project: pysim
Gerrit-Branch: master
Gerrit-Change-Id: I69e104f1d78165c12291317326dbab05977a1574
Gerrit-Change-Number: 36961
Gerrit-PatchSet: 3
Gerrit-Owner: laforge <laforge(a)osmocom.org>
Gerrit-Reviewer: Jenkins Builder
Gerrit-Attention: laforge <laforge(a)osmocom.org>
Gerrit-MessageType: newpatchset