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