diff --git a/docs/pages/configuration/directives.mdx b/docs/pages/configuration/directives.mdx index 02891a9..9da3812 100644 --- a/docs/pages/configuration/directives.mdx +++ b/docs/pages/configuration/directives.mdx @@ -166,6 +166,7 @@ your-directive: - condition: null command: show ip route {target} field: + description: IP of target validation: '[0-9a-f\.\:]+' ``` @@ -178,6 +179,7 @@ your-directive: - condition: null command: show ip bgp community {target} field: + description: BGP community to show options: - value: "65001:1" description: Provider A Routes diff --git a/hyperglass/models/api/query.py b/hyperglass/models/api/query.py index 3b941fa..a8791bc 100644 --- a/hyperglass/models/api/query.py +++ b/hyperglass/models/api/query.py @@ -7,7 +7,8 @@ import secrets from datetime import datetime # Third Party -from pydantic import Field, BaseModel, ConfigDict, field_validator +from pydantic import BaseModel, ConfigDict, field_validator, StringConstraints +from typing_extensions import Annotated # Project from hyperglass.log import log @@ -21,6 +22,11 @@ from hyperglass.exceptions.private import InputValidationError from ..config.devices import Device +QueryLocation = Annotated[str, StringConstraints(strict=True, min_length=1, strip_whitespace=True)] +QueryTarget = Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)] +QueryType = Annotated[str, StringConstraints(strict=True, min_length=1, strip_whitespace=True)] + + class SimpleQuery(BaseModel): """A simple representation of a post-validated query.""" @@ -39,12 +45,12 @@ class Query(BaseModel): model_config = ConfigDict(extra="allow", alias_generator=snake_to_camel, populate_by_name=True) # Device `name` field - query_location: str = Field(strict=True, min_length=1, strip_whitespace=True) + query_location: QueryLocation - query_target: t.Union[t.List[str], str] = Field(min_length=1, strip_whitespace=True) + query_target: t.Union[t.List[QueryTarget], QueryTarget] # Directive `id` field - query_type: str = Field(strict=True, min_length=1, strip_whitespace=True) + query_type: QueryType _kwargs: t.Dict[str, t.Any] def __init__(self, **data) -> None: diff --git a/hyperglass/models/config/devices.py b/hyperglass/models/config/devices.py index 99d868b..70cac21 100644 --- a/hyperglass/models/config/devices.py +++ b/hyperglass/models/config/devices.py @@ -170,7 +170,7 @@ class Device(HyperglassModelWithId, extra="allow"): @field_validator("address") def validate_address( - cls, value: t.Union[IPv4Address, IPv6Address, str], values: t.Dict[str, t.Any] + cls, value: t.Union[IPv4Address, IPv6Address, str], info: ValidationInfo ) -> t.Union[IPv4Address, IPv6Address, str]: """Ensure a hostname is resolvable.""" @@ -178,14 +178,14 @@ class Device(HyperglassModelWithId, extra="allow"): if not any(resolve_hostname(value)): raise ConfigError( "Device '{d}' has an address of '{a}', which is not resolvable.", - d=values["name"], + d=info.data["name"], a=value, ) return value @field_validator("avatar") def validate_avatar( - cls, value: t.Union[FilePath, None], values: t.Dict[str, t.Any] + cls, value: t.Union[FilePath, None], info: ValidationInfo ) -> t.Union[FilePath, None]: """Migrate avatar to static directory.""" if value is not None: @@ -198,7 +198,7 @@ class Device(HyperglassModelWithId, extra="allow"): target = Settings.static_path / "images" / value.name copied = shutil.copy2(value, target) log.bind( - device=values["name"], + device=info.data["name"], source=str(value), destination=str(target), ).debug("Copied device avatar") @@ -210,24 +210,24 @@ class Device(HyperglassModelWithId, extra="allow"): return value @field_validator("platform", mode="before") - def validate_platform(cls: "Device", value: t.Any, values: t.Dict[str, t.Any]) -> str: + def validate_platform(cls: "Device", value: t.Any, info: ValidationInfo) -> str: """Validate & rewrite device platform, set default `directives`.""" if value == "http": - if values.get("http") is None: + if info.data.get("http") is None: raise ConfigError( "Device '{device}' has platform 'http' configured, but no http parameters are defined.", - device=values["name"], + device=info.data["name"], ) if value is None: - if values.get("http") is not None: + if info.data.get("http") is not None: value = "http" else: # Ensure device platform is defined. raise ConfigError( "Device '{device}' is missing a 'platform' (Network Operating System) property", - device=values["name"], + device=info.data["name"], ) if value in SCRAPE_HELPERS.keys():