Commit de5ab581 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

NBI - IETF ACL connector: 

- Cosmetic updates
parent 2e27b0c8

src/nbi/service/rest_server/nbi_plugins/ietf_acl/ietf_acl_parser.py

+44 −10
Original line number Diff line number Diff line
@@ -12,9 +12,8 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



from typing import List, Dict, Optional, TypedDict

from typing import List, Dict

from pydantic import BaseModel, Field



from common.proto.acl_pb2 import AclForwardActionEnum, AclRuleTypeEnum, AclEntry

from common.proto.context_pb2 import ConfigActionEnum, ConfigRule
@@ -139,7 +138,7 @@ def config_rule_from_ietf_acl( 
    acl_entry.match.src_port = source_port

    acl_entry.match.dst_port = destination_port

    acl_entry.match.dscp = acl_ip_data["dscp"]

    acl_entry.match.flags = acl_tcp_data["flags"]

    acl_entry.match.tcp_flags = acl_tcp_data["flags"]

    acl_entry.action.forward_action = getattr(AclForwardActionEnum, IETF_TFS_FORWARDING_ACTION_MAPPING[ietf_action])

    acl_rule_set.entries.append(acl_entry)
@@ -150,15 +149,50 @@ def ietf_acl_from_config_rule_resource_value(config_rule_rv: Dict) -> Dict: 
    acl_entry = rule_set['entries'][0]

    match_ = acl_entry['match']



    ipv4 = Ipv4(dscp=match_["dscp"], source_ipv4_network=match_["src_address"], destination_ipv4_network=match_["dst_address"])

    tcp = Tcp(flags=match_["flags"], source_port=Port(port=match_["src_port"]), destination_port=Port(port=match_["dst_port"]))

    ipv4 = Ipv4(

        dscp=match_["dscp"],

        source_ipv4_network=match_["src_address"],

        destination_ipv4_network=match_["dst_address"]

    )

    tcp = Tcp(

        flags=match_["tcp_flags"],

        source_port=Port(port=match_["src_port"]),

        destination_port=Port(port=match_["dst_port"])

    )

    matches = Matches(ipvr=ipv4, tcp=tcp)

    aces = Aces(ace=[Ace(matches=matches, actions=Action(forwarding=TFS_IETF_FORWARDING_ACTION_MAPPING[acl_entry["action"]["forward_action"]]))])

    acl = Acl(name=rule_set["name"], type=TFS_IETF_RULE_TYPE_MAPPING[rule_set["type"]], aces=aces)

    acl_sets = AclSets(acl_sets=AclSet(acl_set=[Name(name=rule_set["name"])]))

    aces = Aces(ace=[

        Ace(

            matches=matches,

            actions=Action(

                forwarding=TFS_IETF_FORWARDING_ACTION_MAPPING[acl_entry["action"]["forward_action"]]

            )

        )

    ])

    acl = Acl(

        name=rule_set["name"],

        type=TFS_IETF_RULE_TYPE_MAPPING[rule_set["type"]],

        aces=aces

    )

    acl_sets = AclSets(

        acl_sets=AclSet(

            acl_set=[

                Name(name=rule_set["name"])

            ]

        )

    )

    ingress = Ingress(ingress=acl_sets)

    interfaces = Interfaces(interface=[Interface(interface_id=config_rule_rv["interface"], ingress=ingress)])

    acls = Acls(acl=[acl], attachment_points=AttachmentPoints(attachment_points=interfaces))

    interfaces = Interfaces(interface=[

        Interface(

            interface_id=config_rule_rv["interface"],

            ingress=ingress

        )

    ])

    acls = Acls(

        acl=[acl],

        attachment_points=AttachmentPoints(

            attachment_points=interfaces

        )

    )

    ietf_acl = IETF_ACL(acls=acls)



    return ietf_acl.model_dump(by_alias=True)