361 lines
15 KiB
Python
361 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.conf_models import (
|
|
ConfBaseConfig,
|
|
ConfBuiltin,
|
|
ConfGroup,
|
|
ConfInlineRule,
|
|
ConfModule,
|
|
ConfPolicyOverride,
|
|
ConfProfileConfig,
|
|
ConfSelector,
|
|
ConfSource,
|
|
)
|
|
|
|
_ENV_PATTERN = re.compile(r"\$\{([A-Z0-9_]+)\}")
|
|
_SCALAR_KEYS = {
|
|
"listen",
|
|
"output_dir",
|
|
"cache_dir",
|
|
"mode",
|
|
"allow_lan",
|
|
"log_level",
|
|
"ipv6",
|
|
"append_userinfo_header",
|
|
"userinfo_source_policy",
|
|
}
|
|
_BOOL_KEYS = {"allow_lan", "ipv6", "append_userinfo_header"}
|
|
_MULTI_KEYS = {"source", "selector", "group", "module", "builtin"}
|
|
_SOURCE_TYPES = {"url", "file", "inline", "base64"}
|
|
_PROFILE_SCALAR_KEYS = {
|
|
"name",
|
|
"enabled",
|
|
"sources",
|
|
"include_modules",
|
|
"exclude_modules",
|
|
"include_builtins",
|
|
}
|
|
_PROFILE_MULTI_KEYS = {"override_policy", "prepend_rule", "append_rule"}
|
|
|
|
|
|
class ConfConfigError(ValueError):
|
|
def __init__(self, path: Path, line_no: int, message: str) -> None:
|
|
self.path = path
|
|
self.line_no = line_no
|
|
self.message = message
|
|
super().__init__(f"{path}:{line_no}: {message}")
|
|
|
|
|
|
def _expand_env(value: str) -> str:
|
|
return _ENV_PATTERN.sub(lambda match: os.getenv(match.group(1), ""), value)
|
|
|
|
|
|
def _parse_bool(value: str, *, path: Path, line_no: int, key: str) -> bool:
|
|
normalized = value.strip().lower()
|
|
mapping = {
|
|
"true": True,
|
|
"yes": True,
|
|
"1": True,
|
|
"false": False,
|
|
"no": False,
|
|
"0": False,
|
|
}
|
|
if normalized not in mapping:
|
|
raise ConfConfigError(path, line_no, f"invalid boolean for {key}: {value}")
|
|
return mapping[normalized]
|
|
|
|
|
|
def _parse_csv_fields(value: str, *, path: Path, line_no: int) -> list[str]:
|
|
try:
|
|
row = next(csv.reader([value], skipinitialspace=True))
|
|
except Exception as exc: # noqa: BLE001
|
|
raise ConfConfigError(path, line_no, f"invalid csv payload: {exc}") from exc
|
|
return [field.strip() for field in row]
|
|
|
|
|
|
def _split_key_value(raw_line: str, *, path: Path, line_no: int) -> tuple[str, str]:
|
|
if "=" not in raw_line:
|
|
raise ConfConfigError(path, line_no, "expected key = value")
|
|
key, value = [part.strip() for part in raw_line.split("=", 1)]
|
|
if not key:
|
|
raise ConfConfigError(path, line_no, "key cannot be empty")
|
|
return key, value
|
|
|
|
|
|
def _iter_conf_lines(path: Path) -> list[tuple[int, str]]:
|
|
records: list[tuple[int, str]] = []
|
|
for line_no, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#") or line.startswith(";"):
|
|
continue
|
|
records.append((line_no, raw_line))
|
|
return records
|
|
|
|
|
|
def _ensure_named_records(records: list[Any], *, key: str, path: Path) -> None:
|
|
seen: dict[str, int] = {}
|
|
for record in records:
|
|
name = getattr(record, key)
|
|
line_no = getattr(record, "line_no")
|
|
previous_line = seen.get(name)
|
|
if previous_line is not None:
|
|
raise ConfConfigError(path, line_no, f"duplicate {record.__class__.__name__} key '{name}', first defined at line {previous_line}")
|
|
seen[name] = line_no
|
|
|
|
|
|
def _parse_source(value: str, *, path: Path, line_no: int) -> ConfSource:
|
|
fields = _parse_csv_fields(_expand_env(value), path=path, line_no=line_no)
|
|
if len(fields) < 3:
|
|
raise ConfConfigError(path, line_no, "source requires at least 3 fields: key,type,value")
|
|
key, source_type, source_value, *extras = fields
|
|
if not key:
|
|
raise ConfConfigError(path, line_no, "source key cannot be empty")
|
|
if source_type not in _SOURCE_TYPES:
|
|
raise ConfConfigError(path, line_no, f"unsupported source type: {source_type}")
|
|
options: dict[str, str] = {}
|
|
enabled = True
|
|
cache_ttl: int | None = None
|
|
for item in extras:
|
|
if "=" not in item:
|
|
raise ConfConfigError(path, line_no, f"source option must be k=v: {item}")
|
|
option_key, option_value = [part.strip() for part in item.split("=", 1)]
|
|
if not option_key:
|
|
raise ConfConfigError(path, line_no, "source option key cannot be empty")
|
|
option_value = _expand_env(option_value)
|
|
if option_key == "enabled":
|
|
enabled = _parse_bool(option_value, path=path, line_no=line_no, key="source.enabled")
|
|
elif option_key == "cache_ttl":
|
|
try:
|
|
cache_ttl = int(option_value)
|
|
except ValueError as exc:
|
|
raise ConfConfigError(path, line_no, f"invalid integer for source.cache_ttl: {option_value}") from exc
|
|
else:
|
|
options[option_key] = option_value
|
|
return ConfSource(
|
|
key=key,
|
|
source_type=source_type,
|
|
value=source_value,
|
|
enabled=enabled,
|
|
cache_ttl=cache_ttl,
|
|
options=options,
|
|
line_no=line_no,
|
|
)
|
|
|
|
|
|
def _parse_selector(value: str, *, path: Path, line_no: int) -> ConfSelector:
|
|
fields = _parse_csv_fields(value, path=path, line_no=line_no)
|
|
if len(fields) != 2:
|
|
raise ConfConfigError(path, line_no, "selector requires exactly 2 fields: key,regex")
|
|
key, regex = fields
|
|
if not key or not regex:
|
|
raise ConfConfigError(path, line_no, "selector key and regex cannot be empty")
|
|
return ConfSelector(key=key, regex=regex, line_no=line_no)
|
|
|
|
|
|
def _parse_group(value: str, *, path: Path, line_no: int) -> ConfGroup:
|
|
segments = [segment.strip() for segment in value.split("`")]
|
|
if len(segments) < 2:
|
|
raise ConfConfigError(path, line_no, "group requires at least name`type")
|
|
name, group_type, *tokens = segments
|
|
if not name or not group_type:
|
|
raise ConfConfigError(path, line_no, "group name and type cannot be empty")
|
|
return ConfGroup(name=name, group_type=group_type, tokens=tokens, raw=value.strip(), line_no=line_no)
|
|
|
|
|
|
def _parse_module(value: str, *, path: Path, line_no: int, config_dir: Path) -> ConfModule:
|
|
fields = _parse_csv_fields(value, path=path, line_no=line_no)
|
|
if len(fields) != 5:
|
|
raise ConfConfigError(path, line_no, "module requires exactly 5 fields: key,path,policy,order,enabled")
|
|
key, module_path, policy, order_text, enabled_text = fields
|
|
if not key or not module_path or not policy:
|
|
raise ConfConfigError(path, line_no, "module key, path, and policy cannot be empty")
|
|
try:
|
|
order = int(order_text)
|
|
except ValueError as exc:
|
|
raise ConfConfigError(path, line_no, f"invalid integer for module.order: {order_text}") from exc
|
|
enabled = _parse_bool(enabled_text, path=path, line_no=line_no, key="module.enabled")
|
|
resolved_path = (config_dir / module_path).resolve()
|
|
if not resolved_path.is_file():
|
|
raise ConfConfigError(path, line_no, f"module path does not exist: {module_path}")
|
|
return ConfModule(key=key, path=module_path, policy=policy, order=order, enabled=enabled, line_no=line_no)
|
|
|
|
|
|
def _parse_builtin(value: str, *, path: Path, line_no: int) -> ConfBuiltin:
|
|
fields = _parse_csv_fields(value, path=path, line_no=line_no)
|
|
if len(fields) != 6:
|
|
raise ConfConfigError(path, line_no, "builtin requires exactly 6 fields: key,type,value,policy,order,enabled")
|
|
key, builtin_type, builtin_value, policy, order_text, enabled_text = fields
|
|
if builtin_type not in {"GEOIP", "FINAL"}:
|
|
raise ConfConfigError(path, line_no, f"unsupported builtin type: {builtin_type}")
|
|
if not key or not policy:
|
|
raise ConfConfigError(path, line_no, "builtin key and policy cannot be empty")
|
|
if builtin_type == "GEOIP" and not builtin_value:
|
|
raise ConfConfigError(path, line_no, "builtin GEOIP requires a value")
|
|
try:
|
|
order = int(order_text)
|
|
except ValueError as exc:
|
|
raise ConfConfigError(path, line_no, f"invalid integer for builtin.order: {order_text}") from exc
|
|
enabled = _parse_bool(enabled_text, path=path, line_no=line_no, key="builtin.enabled")
|
|
return ConfBuiltin(
|
|
key=key,
|
|
builtin_type=builtin_type,
|
|
value=builtin_value,
|
|
policy=policy,
|
|
order=order,
|
|
enabled=enabled,
|
|
line_no=line_no,
|
|
)
|
|
|
|
|
|
def _parse_string_list(value: str, *, path: Path, line_no: int) -> list[str]:
|
|
fields = _parse_csv_fields(_expand_env(value), path=path, line_no=line_no)
|
|
return [field for field in fields if field]
|
|
|
|
|
|
def _parse_override_policy(value: str, *, path: Path, line_no: int) -> ConfPolicyOverride:
|
|
fields = _parse_csv_fields(value, path=path, line_no=line_no)
|
|
if len(fields) != 2:
|
|
raise ConfConfigError(path, line_no, "override_policy requires exactly 2 fields: module_key,new_policy")
|
|
module_key, policy = fields
|
|
if not module_key or not policy:
|
|
raise ConfConfigError(path, line_no, "override_policy module key and policy cannot be empty")
|
|
return ConfPolicyOverride(module_key=module_key, policy=policy, line_no=line_no)
|
|
|
|
|
|
def _parse_inline_rule(value: str, *, path: Path, line_no: int, key: str) -> ConfInlineRule:
|
|
expanded = _expand_env(value).strip()
|
|
if not expanded:
|
|
raise ConfConfigError(path, line_no, f"{key} cannot be empty")
|
|
return ConfInlineRule(value=expanded, line_no=line_no)
|
|
|
|
|
|
def _validate_profile_references(
|
|
profile: ConfProfileConfig,
|
|
base: ConfBaseConfig,
|
|
*,
|
|
path: Path,
|
|
field_lines: dict[str, int],
|
|
) -> None:
|
|
source_keys = {item.key for item in base.sources}
|
|
module_keys = {item.key for item in base.modules}
|
|
builtin_keys = {item.key for item in base.builtins}
|
|
|
|
for source in profile.sources:
|
|
if source not in source_keys:
|
|
raise ConfConfigError(path, field_lines.get("sources", 1), f"profile references unknown source: {source}")
|
|
for module in profile.include_modules:
|
|
if module not in module_keys:
|
|
raise ConfConfigError(
|
|
path,
|
|
field_lines.get("include_modules", 1),
|
|
f"profile references unknown module in include_modules: {module}",
|
|
)
|
|
for module in profile.exclude_modules:
|
|
if module not in module_keys:
|
|
raise ConfConfigError(
|
|
path,
|
|
field_lines.get("exclude_modules", 1),
|
|
f"profile references unknown module in exclude_modules: {module}",
|
|
)
|
|
for builtin in profile.include_builtins:
|
|
if builtin not in builtin_keys:
|
|
raise ConfConfigError(
|
|
path,
|
|
field_lines.get("include_builtins", 1),
|
|
f"profile references unknown builtin in include_builtins: {builtin}",
|
|
)
|
|
for override in profile.override_policies:
|
|
if override.module_key not in module_keys:
|
|
raise ConfConfigError(path, override.line_no, f"override_policy references unknown module: {override.module_key}")
|
|
|
|
|
|
def load_conf_base(path: Path) -> ConfBaseConfig:
|
|
config = ConfBaseConfig()
|
|
scalar_values: dict[str, Any] = {}
|
|
config_dir = path.resolve().parent
|
|
|
|
for line_no, raw_line in _iter_conf_lines(path):
|
|
key, value = _split_key_value(raw_line, path=path, line_no=line_no)
|
|
if key in _SCALAR_KEYS:
|
|
if key in scalar_values:
|
|
raise ConfConfigError(path, line_no, f"duplicate scalar key: {key}")
|
|
expanded_value = _expand_env(value)
|
|
if key in _BOOL_KEYS:
|
|
scalar_values[key] = _parse_bool(expanded_value, path=path, line_no=line_no, key=key)
|
|
else:
|
|
scalar_values[key] = expanded_value
|
|
continue
|
|
if key not in _MULTI_KEYS:
|
|
raise ConfConfigError(path, line_no, f"unknown directive: {key}")
|
|
if key == "source":
|
|
config.sources.append(_parse_source(value, path=path, line_no=line_no))
|
|
elif key == "selector":
|
|
config.selectors.append(_parse_selector(value, path=path, line_no=line_no))
|
|
elif key == "group":
|
|
config.groups.append(_parse_group(value, path=path, line_no=line_no))
|
|
elif key == "module":
|
|
config.modules.append(_parse_module(value, path=path, line_no=line_no, config_dir=config_dir))
|
|
elif key == "builtin":
|
|
config.builtins.append(_parse_builtin(value, path=path, line_no=line_no))
|
|
|
|
for key, parsed_value in scalar_values.items():
|
|
setattr(config, key, parsed_value)
|
|
|
|
_ensure_named_records(config.sources, key="key", path=path)
|
|
_ensure_named_records(config.selectors, key="key", path=path)
|
|
_ensure_named_records(config.groups, key="name", path=path)
|
|
_ensure_named_records(config.modules, key="key", path=path)
|
|
_ensure_named_records(config.builtins, key="key", path=path)
|
|
return config
|
|
|
|
|
|
def load_conf_profile(path: Path, *, base_config: ConfBaseConfig) -> ConfProfileConfig:
|
|
profile = ConfProfileConfig()
|
|
scalar_values: dict[str, Any] = {}
|
|
field_lines: dict[str, int] = {}
|
|
|
|
for line_no, raw_line in _iter_conf_lines(path):
|
|
key, value = _split_key_value(raw_line, path=path, line_no=line_no)
|
|
if key in _PROFILE_SCALAR_KEYS:
|
|
if key in scalar_values:
|
|
raise ConfConfigError(path, line_no, f"duplicate scalar key: {key}")
|
|
field_lines[key] = line_no
|
|
if key == "enabled":
|
|
scalar_values[key] = _parse_bool(_expand_env(value), path=path, line_no=line_no, key=key)
|
|
elif key in {"sources", "include_modules", "exclude_modules", "include_builtins"}:
|
|
scalar_values[key] = _parse_string_list(value, path=path, line_no=line_no)
|
|
else:
|
|
scalar_values[key] = _expand_env(value)
|
|
continue
|
|
if key not in _PROFILE_MULTI_KEYS:
|
|
raise ConfConfigError(path, line_no, f"unknown directive: {key}")
|
|
if key == "override_policy":
|
|
profile.override_policies.append(_parse_override_policy(value, path=path, line_no=line_no))
|
|
elif key == "prepend_rule":
|
|
profile.prepend_rules.append(_parse_inline_rule(value, path=path, line_no=line_no, key=key))
|
|
elif key == "append_rule":
|
|
profile.append_rules.append(_parse_inline_rule(value, path=path, line_no=line_no, key=key))
|
|
|
|
for key, parsed_value in scalar_values.items():
|
|
setattr(profile, key, parsed_value)
|
|
|
|
_validate_profile_references(profile, base_config, path=path, field_lines=field_lines)
|
|
return profile
|
|
|
|
|
|
def load_conf_profiles(directory: Path, *, base_config: ConfBaseConfig) -> dict[str, ConfProfileConfig]:
|
|
profiles: dict[str, ConfProfileConfig] = {}
|
|
for path in sorted(directory.glob("*.conf")):
|
|
profile = load_conf_profile(path, base_config=base_config)
|
|
profile_key = path.stem
|
|
if profile_key in profiles:
|
|
raise ConfConfigError(path, 1, f"duplicate profile file key: {profile_key}")
|
|
profiles[profile_key] = profile
|
|
return profiles
|