pa_api.xmlapi.types.config.address

  1# Given a list of subnets,
  2# Find all NAT rules related to an address in the subnet
  3
  4import string
  5from ipaddress import IPv4Network, IPv6Network, ip_network
  6from typing import TYPE_CHECKING, Optional, Union
  7
  8from pydantic import AliasChoices, AliasPath, Field
  9from pydantic.functional_validators import field_validator, model_validator
 10from typing_extensions import Self
 11
 12from pa_api.xmlapi.types.utils import List, ObjectBaseModel, String, XMLBaseModel
 13
 14IPNetwork = Union[IPv4Network, IPv6Network]
 15
 16
 17def get_ip_network(ip_netmask):
 18    try:
 19        if ip_netmask:
 20            return ip_network(ip_netmask, strict=False)
 21    except Exception:
 22        return None
 23
 24
 25# https://docs.pydantic.dev/latest/concepts/alias/#aliaspath-and-aliaschoices
 26class Address(XMLBaseModel):
 27    name: str = Field(validation_alias="@name")
 28    type: Optional[str] = None
 29    prefix: Optional[str] = None
 30    ip_netmask: Optional[str] = Field(
 31        alias="ip-netmask",
 32        validation_alias=AliasChoices(
 33            AliasPath("ip-netmask", "#text"),
 34            "ip-netmask",
 35        ),
 36        default=None,
 37    )
 38    ip_network: Optional[IPNetwork] = None
 39    ip_range: Optional[str] = Field(alias="ip-range", default=None)
 40    fqdn: Optional[String] = None
 41    tags: List[String] = Field(
 42        validation_alias=AliasPath("tag", "member"),
 43        default_factory=list,
 44    )
 45
 46    @field_validator("tags", mode="before")
 47    @classmethod
 48    def validate_tags(cls, v) -> List[str]:
 49        if not v:
 50            return []
 51        if not isinstance(v, list):
 52            return [v]
 53        return v
 54
 55    @model_validator(mode="after")
 56    def validate_ip_network(self) -> Self:
 57        if self.ip_network is None:
 58            self.ip_network = get_ip_network(self.ip_netmask)
 59        if not isinstance(self.ip_network, (IPv4Network, IPv6Network)):
 60            self.ip_network = None
 61        return self
 62
 63    @model_validator(mode="after")
 64    def validate_type(self) -> Self:
 65        address_type = None
 66        if self.prefix:
 67            address_type = "prefix"
 68        elif self.ip_netmask:
 69            address_type = "ip-netmask"
 70        elif self.ip_range:
 71            address_type = "ip-range"
 72        elif self.fqdn:
 73            address_type = "fqdn"
 74        self.type = address_type
 75        return self
 76
 77
 78# ======================================================
 79# Filtering tools
 80
 81
 82def take_until_marker(it, marker):
 83    for c in it:
 84        if c == marker:
 85            return
 86        yield c
 87
 88
 89def take_word(it):
 90    for c in it:
 91        if c in string.whitespace:
 92            return
 93        yield c
 94
 95
 96WORD_CHARSET = string.ascii_letters + string.digits + "_-"
 97AND = 1
 98OR = 2
 99
100
101def _parse_filter(it):
102    node = []
103    for c in it:
104        if c in WORD_CHARSET:
105            value = c
106            for c in it:
107                if c not in WORD_CHARSET:
108                    if value == "or":
109                        value = OR
110                    elif value == "and":
111                        value = AND
112                    node.append(value)
113                    break
114                value += c
115            else:  # We didn't break => We reached the end of the string
116                node.append(value)
117                return node
118        if c == "(":
119            subnode = _parse_filter(it)
120            node.append(subnode)
121        if c == ")":
122            return node
123        if c in ("'", '"'):
124            value = "".join(take_until_marker(it, c))
125            node.append(value)
126    return node
127
128
129def _dump_raw_node(node):
130    for e in node:
131        if e == AND:
132            yield "and"
133        elif e == OR:
134            yield "or"
135        elif isinstance(e, list):
136            yield f"({' '.join(_dump_raw_node(e))})"
137        else:
138            yield repr(e)
139
140
141def dump_raw_node(node):
142    return " ".join(_dump_raw_node(node))
143
144
145def parse_filter(text: str):
146    it = iter(text)
147    return _parse_filter(it)
148
149
150# ======================================================
151
152if TYPE_CHECKING:
153    from pa_api.xmlapi.clients import Client
154
155
156class DynamicFilter(XMLBaseModel):
157    filter: String
158
159    @property
160    def sanitized_filter(self) -> str:
161        raw_node = parse_filter(self.filter)
162        filter = dump_raw_node(raw_node)
163        return filter
164
165
166class AddressGroup(ObjectBaseModel):
167    __resource_xpath__ = "/config/devices/entry/device-group/entry/address-group/entry"
168
169    name: str = Field(validation_alias="@name")
170    description: String = ""
171    disable_override: Optional[bool] = Field(alias="disable-override", default=None)
172
173    @property
174    def xpath(self):
175        return f"{self.__resource_xpath__}[@name='{self.name}']"
176
177    @property
178    def type(self):
179        if self.static_members and self.dynamic_members is not None:
180            raise Exception("AddressGroup is both dynamic and static")
181        if self.dynamic_members is not None:
182            return "dynamic"
183        return "static"
184
185    static_members: List[String] = Field(alias="static", default_factory=list)
186    dynamic_members: Optional[DynamicFilter] = Field(
187        validation_alias="dynamic", default=None
188    )
189
190    tags: List[String] = Field(
191        validation_alias=AliasPath("tag", "member"), default_factory=list
192    )
193
194    def remove_member(self, client: "Client", member: str):
195        """
196        Remove the member from destination.
197
198        NOTE: Rulebase information is required for panorama
199        """
200        # panorama_rule_xpath = f"/config/devices/entry/vsys/entry/rulebase/security/rules/entry[@uuid='{self.uuid}']"
201        member_xpath = f"{self.xpath}/static/member[text()='{member}']"
202        return client.configuration.delete(member_xpath)
203
204
205# def find_addresses(tree):
206#     # addresses_xml = tree.xpath(".//address/entry")
207#     addresses_xml = tree.xpath("./devices/entry/device-group//address/entry")
208#     address_objects = [Address.from_xml(n) for n in addresses_xml]
209
210#     addresses = []
211#     subnets = []
212#     for a in address_objects:
213#         network = a.ip_network
214#         # We do not consider ip ranges for now
215#         if not network:
216#             continue
217#         if network.prefixlen == network.max_prefixlen:
218#             addresses.append(a)
219#         else:
220#             subnets.append(a)
221#     return addresses, subnets