mirror of
https://github.com/Av0idD4rk/firegex.git
synced 2026-10-23 15:30:59 +00:00
863422f202
Expose 28 tools with shared REST operations, role-scoped bearer authentication, revisions, durable retries, snapshots, jobs and timed rollback. Add Docker verification harnesses and setup/workflow documentation in docs/mcp.md and inst.md. Validated with 330 unit tests, two consecutive 25-case MCP/import runs, and restart recovery; the full upstream regression run remains incomplete.
98 lines
4.3 KiB
Python
98 lines
4.3 KiB
Python
from typing import Literal
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
|
|
class Model(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
|
|
class ToolResult(Model):
|
|
ok: bool
|
|
data: dict | None = None
|
|
error: dict | None = None
|
|
change: dict | None = None
|
|
result: dict | None = None
|
|
|
|
|
|
class WriteOptions(Model):
|
|
request_id: str = Field(min_length=1, max_length=128, description="Unique operation key. Reuse when retrying a timed-out call.")
|
|
if_revision: str | None = Field(default=None, description="Revision returned by get/overview; required when changing existing configuration.")
|
|
dry_run: bool = False
|
|
background: bool = False
|
|
confirm_within_seconds: int | None = Field(default=None, ge=5, le=3600,
|
|
description="Automatically roll back unless fgx_change_confirm arrives before this deadline.")
|
|
|
|
|
|
class AddressSpec(Model):
|
|
address_id: str | None = None
|
|
ip_int: str = Field(description="IPv4/IPv6 address, CIDR or interface name such as eth0")
|
|
port: int = Field(ge=1, le=65535)
|
|
edge: Literal["tcp", "tls", "udp", "quic"] | None = None
|
|
target_port: int | None = Field(default=None, ge=1, le=65535)
|
|
upstream: Literal["same", "tcp", "tls"] | None = None
|
|
proxy_ip: str | None = None
|
|
proxy_port: int | None = Field(default=None, ge=1, le=65535)
|
|
|
|
|
|
class ServiceSpec(Model):
|
|
name: str | None = Field(default=None, min_length=1, max_length=100)
|
|
proto: Literal["tcp", "udp", "tls", "quic", "http"] | None = None
|
|
transport: Literal["proxy", "nfqueue", "external"] | None = None
|
|
addresses: list[AddressSpec] | None = Field(default=None, min_length=1, max_length=100,
|
|
description="Complete replacement address list when supplied; omitted means unchanged.")
|
|
fail_open: bool | None = None
|
|
max_connections: int | None = Field(default=None, ge=0)
|
|
over_limit_forwards: bool | None = None
|
|
first_byte_timeout: int | None = Field(default=None, ge=0)
|
|
tls_cert: str | None = Field(default=None, max_length=65536)
|
|
tls_key: str | None = Field(default=None, max_length=65536)
|
|
|
|
|
|
class PatternSpec(Model):
|
|
regex_id: str | None = None
|
|
pattern: str = Field(min_length=1, max_length=65536)
|
|
direction: Literal["client_to_server", "server_to_client", "both"] = "both"
|
|
case_sensitive: bool = True
|
|
active: bool = True
|
|
|
|
|
|
class FilterSpec(Model):
|
|
kind: Literal["regex", "pyfilter"] | None = None
|
|
name: str | None = Field(default=None, min_length=1, max_length=100)
|
|
active: bool | None = None
|
|
position: int | None = Field(default=None, ge=0)
|
|
patterns: list[PatternSpec] | None = Field(default=None, max_length=500,
|
|
description="Complete replacement pattern list; omitted means unchanged.")
|
|
code: str | None = Field(default=None, max_length=524288)
|
|
functions: dict[str, bool] | None = Field(default=None, description="Patch function enabled states by Python function name.")
|
|
|
|
|
|
class Sample(Model):
|
|
name: str = "sample"
|
|
data: str = Field(max_length=524288)
|
|
encoding: Literal["utf8", "base64", "hex"] = "utf8"
|
|
|
|
|
|
class FirewallRule(Model):
|
|
id: str | None = Field(default=None, description="Stable MCP rule ID from fgx_firewall_get; absent for a new rule.")
|
|
name: str = Field(min_length=1, max_length=100)
|
|
active: bool = True
|
|
proto: Literal["tcp", "udp", "both", "any"] = "tcp"
|
|
table: Literal["filter", "mangle"] = "filter"
|
|
mode: Literal["in", "out", "forward"] = "in"
|
|
src: str = ""
|
|
dst: str = ""
|
|
port_src_from: int = Field(default=1, ge=1, le=65535)
|
|
port_src_to: int = Field(default=65535, ge=1, le=65535)
|
|
port_dst_from: int = Field(default=1, ge=1, le=65535)
|
|
port_dst_to: int = Field(default=65535, ge=1, le=65535)
|
|
action: Literal["accept", "drop", "reject"] = "drop"
|
|
|
|
|
|
class FirewallPatch(Model):
|
|
rules: list[FirewallRule] | None = Field(default=None, max_length=1000,
|
|
description="Complete ordered ruleset; omitted keeps current rules.")
|
|
policy: Literal["accept", "drop", "reject"] | None = None
|
|
settings: dict[Literal["keep_rules", "allow_loopback", "allow_established", "allow_icmp",
|
|
"multicast_dns", "allow_upnp", "drop_invalid", "allow_dhcp", "allow_dnat"], bool] | None = None
|