HTTP

The system provides following Python APIs to inspect and modify HTTP/2 message.

alc.http.get_message(): Returns the current HTTP message object of type alc.http.Request or alc.http.Response. In case HTTP processing is currently not active, none is returned.

The `alc.http.Request` class

class alc.http.Request

Class representing an HTTP request.

method

HTTP method (GET/POST/PUT/…), value of the :method pseudo-header.

This attribute is read-only.

path: HTTP path, value of the :path pseudo-header.

headers

A dict of headers where the keys are the header-names (in lowercase) and the values are the header values, both str. Pseudo-headers are not included.

This attribute is read-only, the dict can not be modified.

body: Body of the request: a str in case of a regular message or a tuple of alc.http.MessagePart in case of a multipart message.

client: The client endpoint, a str formatted as “ip:port”.

server: The server endpoint, a str formatted as “ip:port”.

ingress: A boolean indicating whether this request is ingressing (True) or egressing (False).

str(request)

The string representation of a request contains the headers (including :method and :path pseudo-headers) and body of the request, formatted similar to how a HTTP/1.x message is transmitted.

This can be used for debugging purposes, but should not be used in production scripts to obtain any of the information in the request, as the specialized methods are much more performant.

The `alc.http.Response` class

class alc.http.Response

Class representing an HTTP response.

status

HTTP status (200/404/501/…), value of the :status pseudo-header.

This attribute is read-only.

headers

A dict of headers where the keys are the header-names (in lowercase) and the values are the header values, both str. Pseudo-headers are not included.

This attribute is read-only, the dict can not be modified.

body: Body of the response: a str in case of a regular message or a tuple of alc.http.MessagePart in case of a multipart message.

client: The client endpoint, a str formatted as “ip:port”.

server: The server endpoint, a str formatted as “ip:port”.

ingress: A boolean indicating whether this response is ingressing (True) or egressing (False).

str(response)

The string representation of a request contains the headers (including :method and :path pseudo-headers) and body of the request, formatted similar to how a HTTP/1.x message is transmitted.

This can be used for debugging purposes, but should not be used in production scripts to obtain any of the information in the request, as the specialized methods are much more performant.

The `alc.http.MessagePart` class

class alc.http.MessagePart

Class representing a single part of a multipart request/response.

headers

A dict of headers where the keys are the header-names (in lowercase) and the values are the header values, both str. Pseudo-headers are not included.

This attribute is read-only, the dict can also not be modified.

body(): Body of the message part as a str. Can be modified to be a str or bytes.

str(part)

The string representation of a request contains the headers (including :method and :path pseudo-headers) and body of the request, formatted similar to how a HTTP/1.x message is transmitted.

This can be used for debugging purposes, but should not be used in production scripts to obtain any of the information in the request, as the specialized methods are much more performant.

Examples

Example 1: Check for Request/Response and modify body

from alc import http
import json

msg = http.get_message()

# You can check if the message is a request or response by asserting its type
if isinstance(msg, http.Request):

    # The body can either be singular or multipart,
    # which is easily checked by checking if it's a tuple or a str.
    if 'sm-contexts' in msg.path and isinstance(msg.body, tuple):
        # A tuple is easily iterated over
        for part in msg.body:
            # Using the get-method on a dict will return None in case the header is not there.
            # This method is preferred over using try-catch in combination with indexing the dict
            # using the msg.headers['content-type'] syntax.
            if part.headers.get('content-type') == 'application/json':
                # fetch the body and decode as json
                body = json.loads(part.body)
                # if present, always set selection-mode to VERIFIED in the message from AMF
                if 'selMode' in body:
                    body['selMode'] = 'VERIFIED'
                    # re-encode to json and set the body to the modified json
                part.body = json.dumps(body)

    # The same but for a single body
    elif 'sm-policies' in msg.path and msg.headers.get('content-type') == 'application/json':
        body = json.loads(msg.body)
        if 'ipDomain' in body:
            body['ipDomain'] = 'vprn3'
            # re-encode to json and set the body to the modified json
            msg.body = json.dumps(body)

Example 2: Modify the Request path

from alc import http
import json
import re

regex = re.compile(r"([?&])gpsi=msisdn-\d+&?")

msg = http.get_message()
if type(msg) is http.Request:
    if 'npcf-smpolicycontrol' in msg.path:
        msg.path = regex.sub(r"\1", msg.path)

Example 3: More complex script to add QOS data

from alc import http
import json

# dict containing qos-data descriptions
QOS_DATA = {
    "QoS Data 1": {
        "5qi": 1,
        "qosId": "QoS Data 1",
        "arp": {"preemptCap": "NOT_PREEMPT", "priorityLevel": 9, "preemptVuln": "NOT_PREEMPTABLE"},
        "defQosFlowIndication": False,
        "gbrDl": "2.000 Mbps",
        "maxbrDl": "2.000 Mbps",
        "reflectiveQos": False,
        "gbrUl": "2.000 Mbps",
        "maxbrUl": "2.000 Mbps",
        "qnc": False,
    },
    "QoS Data 2": {
        "5qi": 2,
        "qosId": "QoS Data 2",
        "arp": {"preemptCap": "NOT_PREEMPT", "priorityLevel": 9, "preemptVuln": "NOT_PREEMPTABLE"},
        "defQosFlowIndication": False,
        "gbrDl": "2.000 Mbps",
        "maxbrDl": "2.000 Mbps",
        "reflectiveQos": False,
        "gbrUl": "2.000 Mbps",
        "maxbrUl": "2.000 Mbps",
        "qnc": False,
    }
}


def add_qos_data(body):
    body_decoded = json.loads(body)

    if not isinstance(body_decoded, dict):
        return None  # body is not a json dictionary

    sm_policy_decision = body_decoded.get('smPolicyDecision')
    if sm_policy_decision is None:
        return None  # no smPolicyDecision (e.g. no PCF update-notify) -> skip

    pcc_rules = sm_policy_decision.get('pccRules')
    if pcc_rules is None:
        return None  # no pcc-rules in smPolicyDecision -> skip

    # fetch all "refQosData" that is referred by added pcc-rules in this message
    ref_qos_data = { ref_qos_data for rule in pcc_rules.values() for ref_qos_data in rule.get('refQosData', []) }

    # construct the preprovisioned qos-data for the ones referenced in this message
    added_qos_data = { qos_data_name: QOS_DATA[qos_data_name] for qos_data_name in ref_qos_data if qos_data_name in QOS_DATA }

    if not added_qos_data:
        return None  # nothing to add

    if 'qosDecs' in sm_policy_decision:
        # add the extra qos-data to the existing qosDecs
        sm_policy_decision['qosDecs'].update(added_qos_data)
    else:
        # qos-data didn't exist -> add the extra qosDecs to the smPolicyDecision
        sm_policy_decision['qosDecs'] = added_qos_data

    # return the body with the added info
    return json.dumps(body_decoded)


def add_qos_data_to_msg():
    msg = http.get_message()

    if not isinstance(msg, http.Request):
        return  # no request (can be omitted if python-policy only enables script on http2 request)

    if not msg.method == "POST":
        return  # no POST method -> skip
    if not msg.path.startswith("/npcf-smpolicycontrol/v1/sm-policies"):
        return  # no nPCF -> skip
    if not msg.path.endswith('/update'):
        return  # no update -> skip

    content_type = msg.headers.get('content-type')
    if content_type != 'application/json':
        return  # no json content -> skip
    if msg.body == 'null':
        return  # message is empty -> skip

    if isinstance(msg.body, tuple):
        # multipart body
        result = []
        has_changed = False
        for part in msg.body:
            new_part = add_qos_data(part)
            if new_part is not None:
                result.append(new_part)
                has_changed = True
            else:
                result.append(part)
        if has_changed:
            msg.body = tuple(result)
    else:
        # single body
        result = add_qos_data(msg.body)
        if result is not None:
            msg.body = result


add_qos_data_to_msg()

HTTP

The alc.http.Request class

The alc.http.Response class

The alc.http.MessagePart class

Examples

Example 1: Check for Request/Response and modify body

Example 2: Modify the Request path

Example 3: More complex script to add QOS data

The `alc.http.Request` class

The `alc.http.Response` class

The `alc.http.MessagePart` class