253 lines
9.6 KiB
Python
253 lines
9.6 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from app.conf_models import (
|
|
ConfBaseConfig,
|
|
ConfGroup,
|
|
ConfProfileConfig,
|
|
ConfResolvedBuiltin,
|
|
ConfResolvedGroup,
|
|
ConfResolvedModule,
|
|
ConfResolvedSource,
|
|
ConfRuntimePlan,
|
|
)
|
|
from app.services.conf_loader import ConfConfigError
|
|
from app.services.rules import load_rule_payload
|
|
|
|
_GROUP_TYPES = {"select", "url-test", "fallback", "load-balance"}
|
|
_CIDR_CHARS = set("0123456789abcdefABCDEF:./")
|
|
_SELECT_SPECIAL_MEMBERS = {"@source_auto_groups", "@all_proxies"}
|
|
|
|
|
|
def _parse_group_member(token: str) -> str:
|
|
if token.startswith("[]"):
|
|
return token[2:]
|
|
return token
|
|
|
|
|
|
def _parse_group(group: ConfGroup, *, selectors: dict[str, str], known_group_names: set[str], path: Path) -> ConfResolvedGroup:
|
|
if group.group_type not in _GROUP_TYPES:
|
|
raise ConfConfigError(path, group.line_no, f"unsupported group type: {group.group_type}")
|
|
if group.group_type == "select":
|
|
selector_key: str | None = None
|
|
members = [_parse_group_member(token) for token in group.tokens if token.strip()]
|
|
if not members:
|
|
raise ConfConfigError(path, group.line_no, "select group must define at least one member")
|
|
for token in group.tokens:
|
|
stripped = token.strip()
|
|
if stripped.startswith("@"):
|
|
if stripped in _SELECT_SPECIAL_MEMBERS:
|
|
continue
|
|
selector_key = stripped[1:]
|
|
if selector_key not in selectors:
|
|
raise ConfConfigError(path, group.line_no, f"group references unknown selector: {selector_key}")
|
|
continue
|
|
if stripped.startswith("[]"):
|
|
target = stripped[2:]
|
|
if target not in {"DIRECT", "REJECT"} and target not in known_group_names:
|
|
raise ConfConfigError(path, group.line_no, f"group references unknown group: {target}")
|
|
return ConfResolvedGroup(
|
|
name=group.name,
|
|
group_type=group.group_type,
|
|
members=[] if selector_key else members,
|
|
selector_key=selector_key,
|
|
raw=group.raw,
|
|
)
|
|
|
|
if not group.tokens:
|
|
raise ConfConfigError(path, group.line_no, f"{group.group_type} group requires selector/member arguments")
|
|
|
|
first = group.tokens[0].strip()
|
|
selector_key: str | None = None
|
|
members: list[str] = []
|
|
if first.startswith("@"):
|
|
selector_key = first[1:]
|
|
if selector_key not in selectors:
|
|
raise ConfConfigError(path, group.line_no, f"group references unknown selector: {selector_key}")
|
|
else:
|
|
members.append(_parse_group_member(first))
|
|
if first.startswith("[]"):
|
|
target = first[2:]
|
|
if target not in {"DIRECT", "REJECT"} and target not in known_group_names:
|
|
raise ConfConfigError(path, group.line_no, f"group references unknown group: {target}")
|
|
|
|
url = group.tokens[1].strip() if len(group.tokens) > 1 and group.tokens[1].strip() else None
|
|
interval: int | None = None
|
|
tolerance: int | None = None
|
|
if len(group.tokens) > 2 and group.tokens[2].strip():
|
|
try:
|
|
interval = int(group.tokens[2].split(",", 1)[0].strip())
|
|
except ValueError as exc:
|
|
raise ConfConfigError(path, group.line_no, f"invalid group interval: {group.tokens[2]}") from exc
|
|
if len(group.tokens) > 3 and group.tokens[3].strip():
|
|
tolerance_token = group.tokens[3].split(",")[-1].strip()
|
|
if tolerance_token:
|
|
try:
|
|
tolerance = int(tolerance_token)
|
|
except ValueError as exc:
|
|
raise ConfConfigError(path, group.line_no, f"invalid group tolerance: {group.tokens[3]}") from exc
|
|
|
|
return ConfResolvedGroup(
|
|
name=group.name,
|
|
group_type=group.group_type,
|
|
members=members,
|
|
selector_key=selector_key,
|
|
url=url,
|
|
interval=interval,
|
|
tolerance=tolerance,
|
|
raw=group.raw,
|
|
)
|
|
|
|
|
|
def _resolve_builtin_rule_line(builtin_type: str, value: str, policy: str) -> str:
|
|
if builtin_type == "GEOIP":
|
|
return f"GEOIP,{value},{policy}"
|
|
return f"MATCH,{policy}"
|
|
|
|
|
|
def _render_rule_line(payload_line: str, policy: str) -> str:
|
|
rendered = payload_line.strip()
|
|
if not rendered:
|
|
return ""
|
|
if "," not in rendered:
|
|
if "/" in rendered and set(rendered) <= _CIDR_CHARS:
|
|
prefix = "IP-CIDR6" if ":" in rendered else "IP-CIDR"
|
|
rendered = f"{prefix},{rendered}"
|
|
else:
|
|
rendered = f"DOMAIN-SUFFIX,{rendered}"
|
|
parts = [part.strip() for part in rendered.split(",")]
|
|
if parts and parts[-1] == "no-resolve":
|
|
parts.insert(len(parts) - 1, policy)
|
|
return ",".join(parts)
|
|
return f"{rendered},{policy}"
|
|
|
|
|
|
def build_conf_rule_lines(plan: ConfRuntimePlan, *, base_path: Path) -> list[str]:
|
|
config_dir = base_path.resolve().parent
|
|
lines: list[str] = []
|
|
final_lines: list[str] = []
|
|
|
|
for rule in plan.prepend_rules:
|
|
if rule.startswith("MATCH,"):
|
|
final_lines.append(rule)
|
|
else:
|
|
lines.append(rule)
|
|
|
|
for module in plan.selected_modules:
|
|
module_path = (config_dir / module.path).resolve()
|
|
if not module_path.is_file():
|
|
raise FileNotFoundError(f"Rule file missing: {module.path}")
|
|
for payload_line in load_rule_payload(module_path):
|
|
rendered = _render_rule_line(payload_line, module.policy)
|
|
if rendered:
|
|
if rendered.startswith("MATCH,"):
|
|
final_lines.append(rendered)
|
|
else:
|
|
lines.append(rendered)
|
|
|
|
for builtin in plan.selected_builtins:
|
|
if builtin.rule_line.startswith("MATCH,"):
|
|
final_lines.append(builtin.rule_line)
|
|
else:
|
|
lines.append(builtin.rule_line)
|
|
|
|
for rule in plan.append_rules:
|
|
if rule.startswith("MATCH,"):
|
|
final_lines.append(rule)
|
|
else:
|
|
lines.append(rule)
|
|
|
|
return [*lines, *final_lines]
|
|
|
|
|
|
def resolve_conf_runtime_plan(
|
|
*,
|
|
profile_key: str,
|
|
base_config: ConfBaseConfig,
|
|
profile_config: ConfProfileConfig,
|
|
base_path: Path,
|
|
) -> ConfRuntimePlan:
|
|
if not profile_config.enabled:
|
|
raise ConfConfigError(base_path, 1, f"profile is disabled: {profile_key}")
|
|
|
|
selectors = {item.key: item.regex for item in base_config.selectors}
|
|
source_map = {item.key: item for item in base_config.sources}
|
|
module_map = {item.key: item for item in base_config.modules}
|
|
builtin_map = {item.key: item for item in base_config.builtins}
|
|
override_map = {item.module_key: item.policy for item in profile_config.override_policies}
|
|
|
|
selected_source_keys = profile_config.sources or [item.key for item in base_config.sources if item.enabled]
|
|
selected_sources = [
|
|
ConfResolvedSource(
|
|
key=source_map[key].key,
|
|
display_name=str(source_map[key].options.get("display_name", source_map[key].key)).strip() or source_map[key].key,
|
|
source_type=source_map[key].source_type,
|
|
value=source_map[key].value,
|
|
enabled=source_map[key].enabled,
|
|
cache_ttl=source_map[key].cache_ttl,
|
|
options=dict(source_map[key].options),
|
|
)
|
|
for key in selected_source_keys
|
|
if source_map[key].enabled
|
|
]
|
|
|
|
excluded_modules = set(profile_config.exclude_modules)
|
|
include_keys = profile_config.include_modules or [item.key for item in base_config.modules if item.enabled]
|
|
selected_modules: list[ConfResolvedModule] = []
|
|
for key in include_keys:
|
|
module = module_map[key]
|
|
if not module.enabled or key in excluded_modules:
|
|
continue
|
|
selected_modules.append(
|
|
ConfResolvedModule(
|
|
key=module.key,
|
|
path=module.path,
|
|
policy=override_map.get(module.key, module.policy),
|
|
order=module.order,
|
|
enabled=module.enabled,
|
|
)
|
|
)
|
|
selected_modules.sort(key=lambda item: (item.order, item.key))
|
|
|
|
builtin_keys = profile_config.include_builtins or [item.key for item in base_config.builtins if item.enabled]
|
|
selected_builtins: list[ConfResolvedBuiltin] = []
|
|
for key in builtin_keys:
|
|
builtin = builtin_map[key]
|
|
if not builtin.enabled:
|
|
continue
|
|
selected_builtins.append(
|
|
ConfResolvedBuiltin(
|
|
key=builtin.key,
|
|
rule_line=_resolve_builtin_rule_line(builtin.builtin_type, builtin.value, builtin.policy),
|
|
policy=builtin.policy,
|
|
order=builtin.order,
|
|
enabled=builtin.enabled,
|
|
)
|
|
)
|
|
selected_builtins.sort(key=lambda item: (item.order, item.key))
|
|
|
|
known_group_names = {group.name for group in base_config.groups}
|
|
resolved_groups = [
|
|
_parse_group(group, selectors=selectors, known_group_names=known_group_names, path=base_path)
|
|
for group in base_config.groups
|
|
]
|
|
|
|
return ConfRuntimePlan(
|
|
profile_key=profile_key,
|
|
profile_name=profile_config.name,
|
|
selected_sources=selected_sources,
|
|
selected_modules=selected_modules,
|
|
selected_builtins=selected_builtins,
|
|
selectors=selectors,
|
|
groups=resolved_groups,
|
|
prepend_rules=[item.value for item in profile_config.prepend_rules],
|
|
append_rules=[item.value for item in profile_config.append_rules],
|
|
mode=base_config.mode,
|
|
allow_lan=base_config.allow_lan,
|
|
log_level=base_config.log_level,
|
|
ipv6=base_config.ipv6,
|
|
append_userinfo_header=base_config.append_userinfo_header,
|
|
userinfo_source_policy=base_config.userinfo_source_policy,
|
|
)
|