ietf_acl_client.py

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import requests, time
from typing import Optional
from requests.auth import HTTPBasicAuth

BASE_URL = '{:s}://{:s}:{:d}/restconf/data'
ACLS_URL = '{:s}/device={:s}/ietf-access-control-list:acls'
ACL_URL  = '{:s}/device={:s}/ietf-access-control-list:acl={:s}'

CSG1_DEVICE_UUID = '118295c8-318a-52ec-a394-529fc4b70f2f' # router: 128.32.10.1
ACL_NAME         = 'sample-ipv4-acl'
ACL_RULE         = {"ietf-access-control-list:acls": {
    "acl": [{
        "name": "sample-ipv4-acl", "type": "ipv4-acl-type",
        "aces": {"ace": [{
            "name": "rule1",
            "matches": {
                "ipv4": {
                    "source-ipv4-network": "128.32.10.6/24",
                    "destination-ipv4-network": "172.10.33.0/24",
                    "dscp": 18
                },
                "tcp": {
                    "source-port": {"operator": "eq", "port": 1444},
                    "destination-port": {"operator": "eq", "port": 1333},
                    "flags": "syn"
                }
            },
            "actions": {"forwarding": "drop"}
        }]}
    }],
    "attachment-points": {"interface": [{
        "interface-id": "200",
        "ingress": {"acl-sets": {"acl-set": [{"name": "sample-ipv4-acl"}]}}
    }]
}}}

class TfsIetfAclClient:
    def __init__(
        self, host : str = 'localhost', port : int = 80, schema : str = 'http',
        username : Optional[str] = 'admin', password : Optional[str] = 'admin',
        timeout : int = 10, allow_redirects : bool = True, verify : bool = False
    ) -> None:
        self._base_url = BASE_URL.format(schema, host, port)
        auth = HTTPBasicAuth(username, password) if username is not None and password is not None else None
        self._settings = dict(auth=auth, timeout=timeout, allow_redirects=allow_redirects, verify=verify)

    def post(self, device_uuid : str, ietf_acl_data : dict) -> str:
        request_url = ACLS_URL.format(self._base_url, device_uuid)
        reply = requests.post(request_url, json=ietf_acl_data, **(self._settings))
        return reply.text

    def get(self, device_uuid : str, acl_name : str) -> str:
        request_url = ACL_URL.format(self._base_url, device_uuid, acl_name)
        reply = requests.get(request_url, **(self._settings))
        return reply.text

    def delete(self, device_uuid : str, acl_name : str) -> str:
        request_url = ACL_URL.format(self._base_url, device_uuid, acl_name)
        reply = requests.delete(request_url, **(self._settings))
        return reply.text

def main():
    client = TfsIetfAclClient()
    print(f'ACL rule: {ACL_RULE}')
    post_response = client.post(CSG1_DEVICE_UUID, ACL_RULE)
    print(f'post response: {post_response}')
    time.sleep(.5)
    get_response = client.get(CSG1_DEVICE_UUID, ACL_NAME)
    print(f'get response: {get_response}')
    time.sleep(.5)
    delete_response = client.delete(CSG1_DEVICE_UUID, ACL_NAME)
    print(f'delete response: {delete_response}')

if __name__ == '__main__':
    main()