catarrh/pysim
1
0
Fork 0
mirror of https://gitea.osmocom.org/sim-card/pysim.git synced 2026-03-19 03:48:34 +03:00
Code Issues Packages Projects Releases Wiki Activity

Revert "esim/http_json_api: extend JSON API with server functionality"

Browse Source 
This reverts commit e00c0becca.
This commit is contained in:
Neels Hofmeyr
2026-03-03 23:49:53 +01:00
parent 7398327e1c
commit ecfcf6057a
3 changed files with 49 additions and 321 deletions
Download Patch File Download Diff File Expand all files Collapse all files

215
pySim/esim/http_json_api.py
View File

@@ -21,8 +21,6 @@ import logging
import json
from typing import Optional
import base64
from twisted.web.server import Request
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
@@ -133,16 +131,6 @@ class JsonResponseHeader(ApiParam):
if status not in ['Executed-Success', 'Executed-WithWarning', 'Failed', 'Expired']:
raise ValueError('Unknown/unspecified status "%s"' % status)
class JsonRequestHeader(ApiParam):
"""SGP.22 section 6.5.1.3."""
@classmethod
def verify_decoded(cls, data):
func_req_id = data.get('functionRequesterIdentifier')
if not func_req_id:
raise ValueError('Missing mandatory functionRequesterIdentifier in header')
func_call_id = data.get('functionCallIdentifier')
if not func_call_id:
raise ValueError('Missing mandatory functionCallIdentifier in header')
class HttpStatusError(Exception):
pass
@@ -173,46 +161,36 @@ class ApiError(Exception):
class JsonHttpApiFunction(abc.ABC):
"""Base class for representing an HTTP[s] API Function."""
# The below class variables are used to describe the properties of the API function. Derived classes are expected
# to orverride those class properties with useful values. The prefixes "input_" and "output_" refer to the API
# function from an abstract point of view. Seen from the client perspective, "input_" will refer to parameters the
# client sends to a HTTP server. Seen from the server perspective, "input_" will refer to parameters the server
# receives from the a requesting client. The same applies vice versa to class variables that have an "output_"
# prefix.
# the below class variables are expected to be overridden in derived classes
# path of the API function (e.g. '/gsma/rsp2/es2plus/confirmOrder')
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 = []
# list of mandatory output parameters (for failed response)
output_mandatory_failed = []
# 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'
# additional custom HTTP headers (client requests)
extra_http_req_headers = {}
# additional custom HTTP headers (server responses)
extra_http_res_headers = {}
def __init__(self, url_prefix: str, func_req_id: Optional[str], session: requests.Session):
self.url_prefix = url_prefix
self.func_req_id = func_req_id
self.session = session
def encode_client(self, data: dict) -> dict:
def encode(self, data: dict, func_call_id: Optional[str] = None) -> dict:
"""Validate an encode input dict into JSON-serializable dict for request body."""
output = {}
if func_call_id:
output['header'] = {
'functionRequesterIdentifier': self.func_req_id,
'functionCallIdentifier': func_call_id
}
for p in self.input_mandatory:
if not p in data:
@@ -226,19 +204,22 @@ class JsonHttpApiFunction(abc.ABC):
output[p] = p_class.encode(v)
return output
def decode_client(self, data: dict) -> dict:
def decode(self, data: dict) -> dict:
"""[further] Decode and validate the JSON-Dict of the response body."""
output = {}
output_mandatory = self.output_mandatory
if 'header' in self.output_params:
# 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'])
# In case a provided header (may be optional) indicates that the API function call was unsuccessful, a
# different set of mandatory parameters applies.
header = data.get('header')
if header:
if data['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
output_mandatory = self.output_mandatory_failed
for p in output_mandatory:
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():
@@ -250,160 +231,30 @@ class JsonHttpApiFunction(abc.ABC):
output[p] = p_class.decode(v)
return output
def encode_server(self, data: dict) -> dict:
"""Validate an encode input dict into JSON-serializable dict for response body."""
output = {}
output_mandatory = self.output_mandatory
# In case a provided header (may be optional) indicates that the API function call was unsuccessful, a
# different set of mandatory parameters applies.
header = data.get('header')
if header:
if data['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
output_mandatory = self.output_mandatory_failed
for p in output_mandatory:
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.encode(v)
return output
def decode_server(self, data: dict) -> dict:
"""[further] Decode and validate the JSON-Dict of the request body."""
output = {}
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.decode(v)
return output
class JsonHttpApiClient():
def __init__(self, api_func: JsonHttpApiFunction, url_prefix: str, func_req_id: Optional[str],
session: requests.Session):
self.api_func = api_func
self.url_prefix = url_prefix
self.func_req_id = func_req_id
self.session = session
def call(self, data: dict, func_call_id: Optional[str] = None, timeout=10) -> Optional[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."""
# In case a function caller ID is supplied, use it together with the stored function requestor ID to generate
# and prepend the header field according to SGP.22, section 6.5.1.1 and 6.5.1.3. (the presence of the header
# field is checked by the encode_client method)
if func_call_id:
data = {'header' : {'functionRequesterIdentifier': self.func_req_id,
'functionCallIdentifier': func_call_id}} | data
# Encode the message (the presence of mandatory fields is checked during encoding)
encoded = json.dumps(self.api_func.encode_client(data))
# Apply HTTP request headers according to SGP.22, section 6.5.1
"""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))
req_headers = {
'Content-Type': 'application/json',
'X-Admin-Protocol': 'gsma/rsp/v2.5.0',
}
req_headers.update(self.api_func.extra_http_req_headers)
req_headers.update(self.extra_http_req_headers)
# Perform HTTP request
url = self.url_prefix + self.api_func.path
logger.debug("HTTP REQ %s - hdr: %s '%s'" % (url, req_headers, encoded))
response = self.session.request(self.api_func.http_method, url, data=encoded, headers=req_headers, timeout=timeout)
response = self.session.request(self.http_method, url, data=encoded, headers=req_headers, timeout=timeout)
logger.debug("HTTP RSP-STS: [%u] hdr: %s" % (response.status_code, response.headers))
logger.debug("HTTP RSP: %s" % (response.content))
# Check HTTP response status code and make sure that the returned HTTP headers look plausible (according to
# SGP.22, section 6.5.1)
if response.status_code != self.api_func.expected_http_status:
if response.status_code != self.expected_http_status:
raise HttpStatusError(response)
if not response.headers.get('Content-Type').startswith(req_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)
# Decode response and return the result back to the caller
if response.content:
output = self.api_func.decode_client(response.json())
# In case the response contains a header, check it to make sure that the API call was executed successfully
# (the presence of the header field is checked by the decode_client method)
if 'header' in output:
if output['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
raise ApiError(output['header']['functionExecutionStatus'])
return output
return self.decode(response.json())
return None
class JsonHttpApiServer():
def __init__(self, api_func: JsonHttpApiFunction, call_handler = None):
"""
Args:
api_func : API function definition (JsonHttpApiFunction)
call_handler : handler function to process the request. This function must accept the
decoded request as a dictionary. The handler function must return a tuple consisting
of the response in the form of a dictionary (may be empty), and a function execution
status string ('Executed-Success', 'Executed-WithWarning', 'Failed' or 'Expired')
"""
self.api_func = api_func
if call_handler:
self.call_handler = call_handler
else:
self.call_handler = self.default_handler
def default_handler(self, data: dict) -> (dict, str):
"""default handler, used in case no call handler is provided."""
logger.error("no handler function for request: %s" % str(data))
return {}, 'Failed'
def call(self, request: Request) -> str:
""" Process an incoming request.
Args:
request : request object as received using twisted.web.server
Returns:
encoded JSON string (HTTP response code and headers are set by calling the appropriate methods on the
provided the request object)
"""
# Make sure the request is done with the correct HTTP method
if (request.method.decode() != self.api_func.http_method):
raise ValueError('Wrong HTTP method %s!=%s' % (request.method.decode(), self.api_func.http_method))
# Decode the request
decoded_request = self.api_func.decode_server(json.loads(request.content.read()))
# Run call handler (see above)
data, fe_status = self.call_handler(decoded_request)
# In case a function execution status is returned, use it to generate and prepend the header field according to
# SGP.22, section 6.5.1.2 and 6.5.1.4 (the presence of the header filed is checked by the encode_server method)
if fe_status:
data = {'header' : {'functionExecutionStatus': {'status' : fe_status}}} | data
# Encode the message (the presence of mandatory fields is checked during encoding)
encoded = json.dumps(self.api_func.encode_server(data))
# Apply HTTP request headers according to SGP.22, section 6.5.1
res_headers = {
'Content-Type': 'application/json',
'X-Admin-Protocol': 'gsma/rsp/v2.5.0',
}
res_headers.update(self.api_func.extra_http_res_headers)
for header, value in res_headers.items():
request.setHeader(header, value)
request.setResponseCode(self.api_func.expected_http_status)
# Return the encoded result back to the caller for sending (using twisted/klein)
return encoded