123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import re
|
|
|
|
import yaml
|
|
|
|
from app.config import get_settings
|
|
from app.models import ClientConfig, ResolvedProfile, RuleConfig
|
|
|
|
_CIDR_PATTERN = re.compile(r"^[0-9a-fA-F:.]+/\d+$")
|
|
|
|
|
|
def resolve_policy(policy: str, client: ClientConfig) -> str:
|
|
return (
|
|
policy.replace("{{ main_policy }}", client.main_policy)
|
|
.replace("{{main_policy}}", client.main_policy)
|
|
.replace("{{ direct_policy }}", client.direct_policy)
|
|
.replace("{{direct_policy}}", client.direct_policy)
|
|
)
|
|
|
|
|
|
def load_rule_text(path: Path) -> str:
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
|
|
def load_rule_payload(path: Path) -> list[str]:
|
|
if path.suffix.lower() in {".yaml", ".yml"}:
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
|
payload = data.get("payload", [])
|
|
if not isinstance(payload, list):
|
|
raise ValueError(f"Rule file {path.name} must contain a list field named 'payload'")
|
|
return [str(item).strip() for item in payload if str(item).strip()]
|
|
|
|
lines: list[str] = []
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
lines.append(stripped)
|
|
return lines
|
|
|
|
|
|
def _render_payload_line(payload_line: str, behavior: str) -> str:
|
|
if "," in payload_line:
|
|
return payload_line
|
|
if behavior == "classical":
|
|
if _CIDR_PATTERN.fullmatch(payload_line):
|
|
prefix = "IP-CIDR6" if ":" in payload_line else "IP-CIDR"
|
|
return f"{prefix},{payload_line}"
|
|
return f"DOMAIN-SUFFIX,{payload_line}"
|
|
if behavior == "ipcidr":
|
|
return f"IP-CIDR,{payload_line}"
|
|
if behavior == "domain":
|
|
return f"DOMAIN-SUFFIX,{payload_line}"
|
|
return payload_line
|
|
|
|
|
|
def _attach_policy(rendered_line: str, target: str, append_no_resolve: bool) -> str:
|
|
parts = [part.strip() for part in rendered_line.split(",")]
|
|
if parts and parts[-1] == "no-resolve":
|
|
parts.insert(len(parts) - 1, target)
|
|
return ",".join(parts)
|
|
line = f"{rendered_line},{target}"
|
|
if append_no_resolve:
|
|
line += ",no-resolve"
|
|
return line
|
|
|
|
|
|
def iter_resolved_rule_lines(
|
|
*,
|
|
resolved_profile: ResolvedProfile,
|
|
include_rule_set_references: bool,
|
|
inline_file_payloads: bool,
|
|
) -> list[str]:
|
|
settings = get_settings()
|
|
lines: list[str] = []
|
|
|
|
for rule_name, rule in resolved_profile.rules.items():
|
|
target = resolve_policy(rule.policy, resolved_profile.client)
|
|
for payload_line in rule.payload:
|
|
lines.append(_attach_policy(_render_payload_line(payload_line, rule.behavior), target, rule.no_resolve))
|
|
|
|
if not rule.file:
|
|
continue
|
|
|
|
if include_rule_set_references:
|
|
ref_line = f"RULE-SET,{rule_name},{target}"
|
|
if rule.no_resolve:
|
|
ref_line += ",no-resolve"
|
|
lines.append(ref_line)
|
|
|
|
if not inline_file_payloads:
|
|
continue
|
|
|
|
path = (settings.rules_dir / rule.file).resolve()
|
|
if not path.is_file() or settings.rules_dir.resolve() not in path.parents:
|
|
raise FileNotFoundError(f"Rule file missing: {rule.file}")
|
|
for payload_line in load_rule_payload(path):
|
|
lines.append(_attach_policy(_render_payload_line(payload_line, rule.behavior), target, rule.no_resolve))
|
|
|
|
lines.append(f"MATCH,{resolved_profile.client.main_policy}")
|
|
return lines
|
|
|
|
|
|
def build_rule_provider_entries(
|
|
*,
|
|
resolved_profile: ResolvedProfile,
|
|
base_url: str,
|
|
public_path: str,
|
|
) -> dict[str, dict]:
|
|
providers: dict[str, dict] = {}
|
|
for name, rule in resolved_profile.rules.items():
|
|
if not rule.file:
|
|
continue
|
|
providers[name] = {
|
|
"behavior": rule.behavior,
|
|
"format": rule.format,
|
|
"url": f"{base_url}/{public_path}/rules/{name}.yaml",
|
|
"interval": rule.interval,
|
|
}
|
|
return providers
|