pa_api.xmlapi.types.statistics

  1from datetime import datetime
  2from typing import Any, List, Literal, Optional, Union
  3
  4from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator
  5
  6from pa_api.xmlapi.utils import el2dict as xml2dict
  7
  8
  9class RuleHit(BaseModel):
 10    name: str = Field(validation_alias=AliasChoices("@name", "name"))
 11    device_group: str = Field(alias="device-group")
 12    rulebase: Literal["pre-rulebase", "post-rulebase"] = Field(alias="rulebase")
 13    type: str = Field(alias="type")
 14    state: Optional[str] = Field(alias="rule-state")
 15    modification: datetime = Field(alias="rule-modification-timestamp")
 16    creation: datetime = Field(alias="rule-creation-timestamp")
 17    all_connected: bool = Field(alias="all-connected")
 18
 19    @field_validator("modification", "creation", mode="before")
 20    @classmethod
 21    def ensure_datetime(cls, v: Any):
 22        if not isinstance(v, int):
 23            v = int(v)
 24        return datetime.fromtimestamp(v)
 25
 26    @staticmethod
 27    def from_tuple(t):
 28        device_group, rulebase, rule_type, xml = t
 29        return RuleHit.model_validate(
 30            {
 31                "device-group": device_group,
 32                "rulebase": rulebase,
 33                "type": rule_type,
 34                **xml2dict(xml)["entry"],
 35            }
 36        )
 37
 38
 39class RuleUse(BaseModel):
 40    model_config = ConfigDict(extra="allow")
 41    name: str = Field(validation_alias=AliasChoices("@name", "name"))
 42    description: Optional[str] = Field(default=None)
 43    uuid: str = Field(validation_alias=AliasChoices("@uuid", "uuid"))
 44    state: Optional[str] = Field(alias="rule-state")
 45    bytes: Optional[int] = Field(default=None)
 46    group_tag: Optional[str] = Field(alias="group-tag", default=None)
 47    tag: Optional[List[str]] = Field(default=None)
 48    disabled: Optional[bool] = Field(default=None)
 49    rule_type: Optional[Literal["interzone", "universal"]] = Field(
 50        alias="rule-type", default=None
 51    )
 52    nat_type: Optional[Literal["ipv4", "ipv6"]] = Field(alias="nat-type", default=None)
 53    modification: datetime = Field(alias="rule-modification-timestamp")
 54    creation: datetime = Field(alias="rule-creation-timestamp")
 55
 56    action: Optional[Union[Literal["allow", "deny", "reset-client"], dict]] = Field(
 57        default=None
 58    )
 59    to_interface: str = Field(alias="to-interface", default=None)
 60    protocol: Optional[Literal["tcp", "udp"]] = Field(default=None)
 61    port: Optional[str] = Field(default=None)  # Can be a port range
 62
 63    to_: Optional[List[str]] = Field(alias="from", default=None)
 64    from_: Optional[List[str]] = Field(alias="to", default=None)
 65    source: Optional[List[str]] = Field(default=None)
 66    destination: Optional[List[str]] = Field(default=None)
 67    source_translation: Optional[List[str]] = Field(
 68        alias="source-translation", default=None
 69    )
 70    destination_translation: Optional[List[str]] = Field(
 71        alias="destination-translation", default=None
 72    )
 73
 74    source_user: Optional[List[str]] = Field(alias="source-user", default=None)
 75    application: Optional[List[str]] = Field(default=None)
 76    category: Optional[List[str]] = Field(default=None)
 77    service: Optional[List[str]] = Field(default=None)
 78
 79    icmp_unreachable: Optional[bool] = Field(alias="icmp-unreachable", default=None)
 80    log_start: Optional[bool] = Field(alias="log-start", default=None)
 81    log_end: Optional[bool] = Field(alias="log-end", default=None)
 82    negate_source: Optional[bool] = Field(alias="negate-source", default=None)
 83    negate_destination: Optional[bool] = Field(alias="negate-destination", default=None)
 84
 85    @field_validator(
 86        "tag",
 87        "to_",
 88        "from_",
 89        "source",
 90        "destination",
 91        "source_translation",
 92        "destination_translation",
 93        "source_user",
 94        "application",
 95        "category",
 96        "service",
 97        mode="before",
 98    )
 99    @classmethod
100    def ensure_membership(cls, v: Any):
101        if v is None:
102            return None
103        members = v.get("member") if isinstance(v, dict) else v
104        if isinstance(members, str):
105            members = [members]
106        return members