riglen
@@ -150,6 +150,7 @@ async def client_profile(client_type: str, request: Request, sources: str | None
|
||||
build_thin_profile(
|
||||
client_type=client_type,
|
||||
app_config=app_config,
|
||||
client=client,
|
||||
selected_source_names=[name for name, _ in source_items],
|
||||
base_url=_base_url(request),
|
||||
public_path=(app_config.public_path or settings.public_path).strip("/"),
|
||||
@@ -198,6 +199,7 @@ async def bundle_profile(
|
||||
build_bundle_profile(
|
||||
client_type=client_type,
|
||||
app_config=app_config,
|
||||
client=client,
|
||||
snapshots=snapshots,
|
||||
)
|
||||
)
|
||||
|
||||
@@ -82,26 +82,6 @@ class ProviderDocument(BaseModel):
|
||||
proxies: list[dict[str, Any]]
|
||||
|
||||
|
||||
class ProxyNode(BaseModel):
|
||||
name: str
|
||||
type: str
|
||||
server: str | None = None
|
||||
port: int | None = None
|
||||
udp: bool = True
|
||||
tags: list[str] = Field(default_factory=list)
|
||||
attrs: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
def to_proxy_dict(self) -> dict[str, Any]:
|
||||
data = {"name": self.name, "type": self.type}
|
||||
if self.server is not None:
|
||||
data["server"] = self.server
|
||||
if self.port is not None:
|
||||
data["port"] = self.port
|
||||
data["udp"] = self.udp
|
||||
data.update(self.attrs)
|
||||
return data
|
||||
|
||||
|
||||
class SubscriptionUserInfo(BaseModel):
|
||||
upload: int | None = None
|
||||
download: int | None = None
|
||||
@@ -130,13 +110,3 @@ class SourceSnapshot(BaseModel):
|
||||
document: ProviderDocument
|
||||
headers: dict[str, str] = Field(default_factory=dict)
|
||||
quota: SubscriptionUserInfo | None = None
|
||||
|
||||
|
||||
class ResolvedProfile(BaseModel):
|
||||
client_type: str
|
||||
client: ClientConfig
|
||||
selected_sources: dict[str, SourceConfig] = Field(default_factory=dict)
|
||||
rules: dict[str, RuleConfig] = Field(default_factory=dict)
|
||||
regions: dict[str, RegionConfig] = Field(default_factory=dict)
|
||||
selector_groups: list[ProxyGroupConfig] = Field(default_factory=list)
|
||||
policy_groups: list[ProxyGroupConfig] = Field(default_factory=list)
|
||||
|
||||
@@ -1,309 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from app.models import ProxyGroupConfig, ResolvedProfile, SourceSnapshot
|
||||
from app.services.proxy_pipeline import source_auto_group_name
|
||||
|
||||
|
||||
def _expand_proxy_tokens(
|
||||
proxies: list[str],
|
||||
*,
|
||||
resolved_profile: ResolvedProfile,
|
||||
source_auto_names: list[str],
|
||||
selector_names: list[str],
|
||||
) -> list[str]:
|
||||
client = resolved_profile.client
|
||||
tokens = {
|
||||
"{{ main_policy }}": [client.main_policy],
|
||||
"{{main_policy}}": [client.main_policy],
|
||||
"{{ source_policy }}": [client.source_policy],
|
||||
"{{source_policy}}": [client.source_policy],
|
||||
"{{ mixed_auto_policy }}": [client.mixed_auto_policy],
|
||||
"{{mixed_auto_policy}}": [client.mixed_auto_policy],
|
||||
"{{ manual_policy }}": [client.manual_policy],
|
||||
"{{manual_policy}}": [client.manual_policy],
|
||||
"{{ direct_policy }}": [client.direct_policy],
|
||||
"{{direct_policy}}": [client.direct_policy],
|
||||
"{{ source_auto_groups }}": source_auto_names,
|
||||
"{{source_auto_groups}}": source_auto_names,
|
||||
"{{ selector_groups }}": selector_names,
|
||||
"{{selector_groups}}": selector_names,
|
||||
}
|
||||
expanded: list[str] = []
|
||||
for item in proxies:
|
||||
expanded.extend(tokens.get(item, [item]))
|
||||
return expanded
|
||||
|
||||
|
||||
def _build_filter_group_for_thin(
|
||||
*,
|
||||
resolved_profile: ResolvedProfile,
|
||||
group: ProxyGroupConfig,
|
||||
selected_source_names: list[str],
|
||||
) -> dict[str, Any]:
|
||||
client = resolved_profile.client
|
||||
built: dict[str, Any] = {"name": group.name, "type": group.type, "filter": group.filter}
|
||||
if group.type == "url-test":
|
||||
built["url"] = str(group.url or client.test_url)
|
||||
built["interval"] = group.interval or client.test_interval
|
||||
if group.tolerance is not None:
|
||||
built["tolerance"] = group.tolerance
|
||||
if resolved_profile.client_type == "mihomo":
|
||||
built["use"] = selected_source_names
|
||||
else:
|
||||
built["include-all"] = True
|
||||
return built
|
||||
|
||||
|
||||
def _build_filter_group_for_bundle(
|
||||
*,
|
||||
resolved_profile: ResolvedProfile,
|
||||
group: ProxyGroupConfig,
|
||||
all_proxy_names: list[str],
|
||||
) -> dict[str, Any]:
|
||||
client = resolved_profile.client
|
||||
matched = [name for name in all_proxy_names if group.filter and re.search(group.filter, name)]
|
||||
built: dict[str, Any] = {"name": group.name, "type": group.type, "proxies": matched or [client.direct_policy]}
|
||||
if group.type == "url-test":
|
||||
built["url"] = str(group.url or client.test_url)
|
||||
built["interval"] = group.interval or client.test_interval
|
||||
if group.tolerance is not None:
|
||||
built["tolerance"] = group.tolerance
|
||||
return built
|
||||
|
||||
|
||||
def _build_custom_policy_groups(
|
||||
*,
|
||||
resolved_profile: ResolvedProfile,
|
||||
source_auto_names: list[str],
|
||||
selector_names: list[str],
|
||||
) -> list[dict[str, Any]]:
|
||||
groups: list[dict[str, Any]] = []
|
||||
client = resolved_profile.client
|
||||
for group in resolved_profile.policy_groups:
|
||||
built: dict[str, Any] = {
|
||||
"name": group.name,
|
||||
"type": group.type,
|
||||
"proxies": _expand_proxy_tokens(
|
||||
group.proxies,
|
||||
resolved_profile=resolved_profile,
|
||||
source_auto_names=source_auto_names,
|
||||
selector_names=selector_names,
|
||||
),
|
||||
}
|
||||
if group.type == "url-test":
|
||||
built["url"] = str(group.url or client.test_url)
|
||||
built["interval"] = group.interval or client.test_interval
|
||||
if group.tolerance is not None:
|
||||
built["tolerance"] = group.tolerance
|
||||
groups.append(built)
|
||||
return groups
|
||||
|
||||
|
||||
def build_thin_groups(
|
||||
*,
|
||||
resolved_profile: ResolvedProfile,
|
||||
selected_source_names: list[str],
|
||||
) -> list[dict[str, Any]]:
|
||||
client = resolved_profile.client
|
||||
groups: list[dict[str, Any]] = []
|
||||
source_auto_names: list[str] = []
|
||||
|
||||
for source_name in selected_source_names:
|
||||
source = resolved_profile.selected_sources[source_name]
|
||||
group_name = source_auto_group_name(source.display_name or source_name)
|
||||
source_auto_names.append(group_name)
|
||||
groups.append(
|
||||
{
|
||||
"name": group_name,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"use": [source_name],
|
||||
}
|
||||
)
|
||||
|
||||
if resolved_profile.client_type == "mihomo":
|
||||
mixed_auto = {
|
||||
"name": client.mixed_auto_policy,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"include-all-providers": True,
|
||||
}
|
||||
manual = {
|
||||
"name": client.manual_policy,
|
||||
"type": "select",
|
||||
"proxies": [client.direct_policy],
|
||||
"include-all-providers": True,
|
||||
}
|
||||
else:
|
||||
mixed_auto = {
|
||||
"name": client.mixed_auto_policy,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"include-all": True,
|
||||
}
|
||||
manual = {
|
||||
"name": client.manual_policy,
|
||||
"type": "select",
|
||||
"proxies": [client.direct_policy],
|
||||
"include-all": True,
|
||||
}
|
||||
|
||||
groups.append(mixed_auto)
|
||||
region_names = [region.name for region in resolved_profile.regions.values()]
|
||||
selector_names = [*region_names, *[selector.name for selector in resolved_profile.selector_groups]]
|
||||
|
||||
groups.append({"name": client.source_policy, "type": "select", "proxies": [client.mixed_auto_policy, *source_auto_names, client.direct_policy]})
|
||||
groups.append(manual)
|
||||
groups.append(
|
||||
{
|
||||
"name": client.main_policy,
|
||||
"type": "select",
|
||||
"proxies": [client.source_policy, client.mixed_auto_policy, *selector_names, client.manual_policy, client.direct_policy],
|
||||
}
|
||||
)
|
||||
groups.extend(
|
||||
_build_custom_policy_groups(
|
||||
resolved_profile=resolved_profile,
|
||||
source_auto_names=source_auto_names,
|
||||
selector_names=selector_names,
|
||||
)
|
||||
)
|
||||
|
||||
for region in resolved_profile.regions.values():
|
||||
group: dict[str, Any] = {
|
||||
"name": region.name,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"filter": region.filter,
|
||||
"tolerance": region.tolerance,
|
||||
}
|
||||
if resolved_profile.client_type == "mihomo":
|
||||
group["include-all-providers"] = True
|
||||
else:
|
||||
group["include-all"] = True
|
||||
groups.append(group)
|
||||
|
||||
for selector in resolved_profile.selector_groups:
|
||||
if selector.filter:
|
||||
groups.append(
|
||||
_build_filter_group_for_thin(
|
||||
resolved_profile=resolved_profile,
|
||||
group=selector,
|
||||
selected_source_names=selected_source_names,
|
||||
)
|
||||
)
|
||||
else:
|
||||
groups.append(
|
||||
{
|
||||
"name": selector.name,
|
||||
"type": selector.type,
|
||||
"proxies": _expand_proxy_tokens(
|
||||
selector.proxies,
|
||||
resolved_profile=resolved_profile,
|
||||
source_auto_names=source_auto_names,
|
||||
selector_names=selector_names,
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
return groups
|
||||
|
||||
|
||||
def build_bundle_groups(
|
||||
*,
|
||||
resolved_profile: ResolvedProfile,
|
||||
snapshots: list[SourceSnapshot],
|
||||
source_proxy_names: dict[str, list[str]],
|
||||
) -> list[dict[str, Any]]:
|
||||
client = resolved_profile.client
|
||||
groups: list[dict[str, Any]] = []
|
||||
source_auto_names: list[str] = []
|
||||
all_proxy_names = [name for names in source_proxy_names.values() for name in names]
|
||||
|
||||
for snapshot in snapshots:
|
||||
group_name = source_auto_group_name(snapshot.display_name)
|
||||
source_auto_names.append(group_name)
|
||||
groups.append(
|
||||
{
|
||||
"name": group_name,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"proxies": source_proxy_names.get(snapshot.name) or [client.direct_policy],
|
||||
}
|
||||
)
|
||||
|
||||
groups.append(
|
||||
{
|
||||
"name": client.mixed_auto_policy,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"proxies": all_proxy_names or [client.direct_policy],
|
||||
}
|
||||
)
|
||||
|
||||
region_names = [region.name for region in resolved_profile.regions.values()]
|
||||
selector_names = [*region_names, *[selector.name for selector in resolved_profile.selector_groups]]
|
||||
|
||||
groups.append({"name": client.source_policy, "type": "select", "proxies": [client.mixed_auto_policy, *source_auto_names, client.direct_policy]})
|
||||
groups.append({"name": client.manual_policy, "type": "select", "proxies": [*all_proxy_names, client.direct_policy] if all_proxy_names else [client.direct_policy]})
|
||||
groups.append(
|
||||
{
|
||||
"name": client.main_policy,
|
||||
"type": "select",
|
||||
"proxies": [client.source_policy, client.mixed_auto_policy, *selector_names, client.manual_policy, client.direct_policy],
|
||||
}
|
||||
)
|
||||
groups.extend(
|
||||
_build_custom_policy_groups(
|
||||
resolved_profile=resolved_profile,
|
||||
source_auto_names=source_auto_names,
|
||||
selector_names=selector_names,
|
||||
)
|
||||
)
|
||||
|
||||
for region in resolved_profile.regions.values():
|
||||
matched = [name for name in all_proxy_names if re.search(region.filter, name)]
|
||||
groups.append(
|
||||
{
|
||||
"name": region.name,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"tolerance": region.tolerance,
|
||||
"proxies": matched or [client.direct_policy],
|
||||
}
|
||||
)
|
||||
|
||||
for selector in resolved_profile.selector_groups:
|
||||
if selector.filter:
|
||||
groups.append(
|
||||
_build_filter_group_for_bundle(
|
||||
resolved_profile=resolved_profile,
|
||||
group=selector,
|
||||
all_proxy_names=all_proxy_names,
|
||||
)
|
||||
)
|
||||
else:
|
||||
groups.append(
|
||||
{
|
||||
"name": selector.name,
|
||||
"type": selector.type,
|
||||
"proxies": _expand_proxy_tokens(
|
||||
selector.proxies,
|
||||
resolved_profile=resolved_profile,
|
||||
source_auto_names=source_auto_names,
|
||||
selector_names=selector_names,
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
return groups
|
||||
@@ -1,21 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from app.models import AppConfig, ResolvedProfile
|
||||
|
||||
|
||||
def resolve_profile(
|
||||
*,
|
||||
app_config: AppConfig,
|
||||
client_type: str,
|
||||
selected_source_names: list[str],
|
||||
) -> ResolvedProfile:
|
||||
client = app_config.clients[client_type]
|
||||
return ResolvedProfile(
|
||||
client_type=client_type,
|
||||
client=client,
|
||||
selected_sources={name: app_config.sources[name] for name in selected_source_names if name in app_config.sources},
|
||||
rules=dict(app_config.rules),
|
||||
regions=dict(app_config.regions),
|
||||
selector_groups=list(app_config.selector_groups),
|
||||
policy_groups=list(app_config.policy_groups),
|
||||
)
|
||||
@@ -1,44 +1,143 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
from app.models import AppConfig, SourceSnapshot
|
||||
from app.services.policy_group_builder import build_bundle_groups, build_thin_groups
|
||||
from app.services.profile_resolver import resolve_profile
|
||||
from app.services.proxy_pipeline import build_bundle_proxy_inventory
|
||||
from app.services.rule_resolver import build_rule_provider_entries, iter_resolved_rule_lines
|
||||
from app.models import AppConfig, ClientConfig, ProxyGroupConfig, SourceSnapshot
|
||||
from app.services.rules import build_inline_rules, build_rule_provider_entries, build_rule_set_references
|
||||
|
||||
|
||||
def dump_yaml(data: dict[str, Any]) -> str:
|
||||
return yaml.safe_dump(data, allow_unicode=True, sort_keys=False, default_flow_style=False)
|
||||
|
||||
|
||||
def _build_profile_header(*, resolved_profile) -> dict[str, Any]:
|
||||
client = resolved_profile.client
|
||||
profile: dict[str, Any] = {"mode": client.mode, "ipv6": client.ipv6}
|
||||
def _expand_proxy_tokens(
|
||||
proxies: list[str],
|
||||
*,
|
||||
client: ClientConfig,
|
||||
source_auto_names: list[str],
|
||||
selector_names: list[str],
|
||||
) -> list[str]:
|
||||
expanded: list[str] = []
|
||||
tokens = {
|
||||
"{{ main_policy }}": [client.main_policy],
|
||||
"{{main_policy}}": [client.main_policy],
|
||||
"{{ source_policy }}": [client.source_policy],
|
||||
"{{source_policy}}": [client.source_policy],
|
||||
"{{ mixed_auto_policy }}": [client.mixed_auto_policy],
|
||||
"{{mixed_auto_policy}}": [client.mixed_auto_policy],
|
||||
"{{ manual_policy }}": [client.manual_policy],
|
||||
"{{manual_policy}}": [client.manual_policy],
|
||||
"{{ direct_policy }}": [client.direct_policy],
|
||||
"{{direct_policy}}": [client.direct_policy],
|
||||
"{{ source_auto_groups }}": source_auto_names,
|
||||
"{{source_auto_groups}}": source_auto_names,
|
||||
"{{ selector_groups }}": selector_names,
|
||||
"{{selector_groups}}": selector_names,
|
||||
}
|
||||
for item in proxies:
|
||||
expanded.extend(tokens.get(item, [item]))
|
||||
return expanded
|
||||
|
||||
|
||||
def _build_thin_filter_group(
|
||||
*,
|
||||
client_type: str,
|
||||
client: ClientConfig,
|
||||
group: ProxyGroupConfig,
|
||||
selected_source_names: list[str],
|
||||
) -> dict[str, Any]:
|
||||
built: dict[str, Any] = {
|
||||
"name": group.name,
|
||||
"type": group.type,
|
||||
"filter": group.filter,
|
||||
}
|
||||
if group.type == "url-test":
|
||||
built["url"] = str(group.url or client.test_url)
|
||||
built["interval"] = group.interval or client.test_interval
|
||||
if group.tolerance is not None:
|
||||
built["tolerance"] = group.tolerance
|
||||
if client_type == "mihomo":
|
||||
built["use"] = selected_source_names
|
||||
else:
|
||||
built["include-all"] = True
|
||||
return built
|
||||
|
||||
|
||||
def _build_bundle_filter_group(
|
||||
*,
|
||||
client: ClientConfig,
|
||||
group: ProxyGroupConfig,
|
||||
all_proxy_names: list[str],
|
||||
) -> dict[str, Any]:
|
||||
matched = [name for name in all_proxy_names if group.filter and re.search(group.filter, name)]
|
||||
built: dict[str, Any] = {
|
||||
"name": group.name,
|
||||
"type": group.type,
|
||||
"proxies": matched or [client.direct_policy],
|
||||
}
|
||||
if group.type == "url-test":
|
||||
built["url"] = str(group.url or client.test_url)
|
||||
built["interval"] = group.interval or client.test_interval
|
||||
if group.tolerance is not None:
|
||||
built["tolerance"] = group.tolerance
|
||||
return built
|
||||
|
||||
|
||||
def _build_custom_policy_groups(
|
||||
*,
|
||||
app_config: AppConfig,
|
||||
client: ClientConfig,
|
||||
source_auto_names: list[str],
|
||||
selector_names: list[str],
|
||||
) -> list[dict[str, Any]]:
|
||||
groups: list[dict[str, Any]] = []
|
||||
for group in app_config.policy_groups:
|
||||
built: dict[str, Any] = {
|
||||
"name": group.name,
|
||||
"type": group.type,
|
||||
"proxies": _expand_proxy_tokens(
|
||||
group.proxies,
|
||||
client=client,
|
||||
source_auto_names=source_auto_names,
|
||||
selector_names=selector_names,
|
||||
),
|
||||
}
|
||||
if group.type == "url-test":
|
||||
built["url"] = str(group.url or client.test_url)
|
||||
built["interval"] = group.interval or client.test_interval
|
||||
if group.tolerance is not None:
|
||||
built["tolerance"] = group.tolerance
|
||||
groups.append(built)
|
||||
return groups
|
||||
|
||||
|
||||
def build_thin_profile(
|
||||
*,
|
||||
client_type: str,
|
||||
app_config: AppConfig,
|
||||
client: ClientConfig,
|
||||
selected_source_names: list[str],
|
||||
base_url: str,
|
||||
public_path: str,
|
||||
) -> dict[str, Any]:
|
||||
profile: dict[str, Any] = {
|
||||
"mode": client.mode,
|
||||
"ipv6": client.ipv6,
|
||||
}
|
||||
if client.log_level:
|
||||
profile["log-level"] = client.log_level
|
||||
if resolved_profile.client_type == "mihomo":
|
||||
if client_type == "mihomo":
|
||||
if client.mixed_port is not None:
|
||||
profile["mixed-port"] = client.mixed_port
|
||||
if client.socks_port is not None:
|
||||
profile["socks-port"] = client.socks_port
|
||||
profile["allow-lan"] = client.allow_lan
|
||||
return profile
|
||||
|
||||
|
||||
def _build_proxy_providers(
|
||||
*,
|
||||
resolved_profile,
|
||||
base_url: str,
|
||||
public_path: str,
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
proxy_providers: dict[str, dict[str, Any]] = {}
|
||||
client = resolved_profile.client
|
||||
for name in resolved_profile.selected_sources:
|
||||
if resolved_profile.client_type == "mihomo":
|
||||
for name in selected_source_names:
|
||||
if client_type == "mihomo":
|
||||
proxy_providers[name] = {
|
||||
"type": "http",
|
||||
"url": f"{base_url}/{public_path}/providers/{name}.yaml",
|
||||
@@ -55,42 +154,10 @@ def _build_proxy_providers(
|
||||
"url": f"{base_url}/{public_path}/providers/{name}.yaml",
|
||||
"interval": client.provider_interval,
|
||||
}
|
||||
return proxy_providers
|
||||
|
||||
|
||||
def build_thin_profile(
|
||||
*,
|
||||
client_type: str,
|
||||
app_config: AppConfig,
|
||||
selected_source_names: list[str],
|
||||
base_url: str,
|
||||
public_path: str,
|
||||
) -> dict[str, Any]:
|
||||
resolved_profile = resolve_profile(
|
||||
app_config=app_config,
|
||||
client_type=client_type,
|
||||
selected_source_names=selected_source_names,
|
||||
)
|
||||
profile = _build_profile_header(resolved_profile=resolved_profile)
|
||||
profile["proxy-providers"] = _build_proxy_providers(
|
||||
resolved_profile=resolved_profile,
|
||||
base_url=base_url,
|
||||
public_path=public_path,
|
||||
)
|
||||
profile["proxy-groups"] = build_thin_groups(
|
||||
resolved_profile=resolved_profile,
|
||||
selected_source_names=selected_source_names,
|
||||
)
|
||||
profile["rule-providers"] = build_rule_provider_entries(
|
||||
resolved_profile=resolved_profile,
|
||||
base_url=base_url,
|
||||
public_path=public_path,
|
||||
)
|
||||
profile["rules"] = iter_resolved_rule_lines(
|
||||
resolved_profile=resolved_profile,
|
||||
include_rule_set_references=True,
|
||||
inline_file_payloads=False,
|
||||
)
|
||||
profile["proxy-providers"] = proxy_providers
|
||||
profile["proxy-groups"] = _build_thin_groups(client_type, app_config, client, selected_source_names)
|
||||
profile["rule-providers"] = build_rule_provider_entries(app_config, client, base_url, public_path)
|
||||
profile["rules"] = build_rule_set_references(app_config, client)
|
||||
return profile
|
||||
|
||||
|
||||
@@ -98,24 +165,275 @@ def build_bundle_profile(
|
||||
*,
|
||||
client_type: str,
|
||||
app_config: AppConfig,
|
||||
client: ClientConfig,
|
||||
snapshots: list[SourceSnapshot],
|
||||
) -> dict[str, Any]:
|
||||
resolved_profile = resolve_profile(
|
||||
app_config=app_config,
|
||||
client_type=client_type,
|
||||
selected_source_names=[snapshot.name for snapshot in snapshots],
|
||||
)
|
||||
profile = _build_profile_header(resolved_profile=resolved_profile)
|
||||
proxy_nodes, source_proxy_names = build_bundle_proxy_inventory(snapshots)
|
||||
profile["proxies"] = [node.to_proxy_dict() for node in proxy_nodes]
|
||||
profile["proxy-groups"] = build_bundle_groups(
|
||||
resolved_profile=resolved_profile,
|
||||
snapshots=snapshots,
|
||||
source_proxy_names=source_proxy_names,
|
||||
)
|
||||
profile["rules"] = iter_resolved_rule_lines(
|
||||
resolved_profile=resolved_profile,
|
||||
include_rule_set_references=False,
|
||||
inline_file_payloads=True,
|
||||
)
|
||||
profile: dict[str, Any] = {
|
||||
"mode": client.mode,
|
||||
"ipv6": client.ipv6,
|
||||
}
|
||||
if client.log_level:
|
||||
profile["log-level"] = client.log_level
|
||||
if client_type == "mihomo":
|
||||
if client.mixed_port is not None:
|
||||
profile["mixed-port"] = client.mixed_port
|
||||
if client.socks_port is not None:
|
||||
profile["socks-port"] = client.socks_port
|
||||
profile["allow-lan"] = client.allow_lan
|
||||
|
||||
proxies: list[dict[str, Any]] = []
|
||||
source_proxy_names: dict[str, list[str]] = {}
|
||||
seen: set[str] = set()
|
||||
for snapshot in snapshots:
|
||||
names: list[str] = []
|
||||
for proxy in snapshot.document.proxies:
|
||||
candidate = dict(proxy)
|
||||
name = str(candidate.get("name", "")).strip()
|
||||
if not name:
|
||||
continue
|
||||
original = name
|
||||
index = 2
|
||||
while name in seen:
|
||||
name = f"{original} #{index}"
|
||||
index += 1
|
||||
seen.add(name)
|
||||
candidate["name"] = name
|
||||
proxies.append(candidate)
|
||||
names.append(name)
|
||||
source_proxy_names[snapshot.name] = names
|
||||
|
||||
profile["proxies"] = proxies
|
||||
profile["proxy-groups"] = _build_bundle_groups(app_config, client, snapshots, source_proxy_names)
|
||||
profile["rules"] = build_inline_rules(app_config, client)
|
||||
return profile
|
||||
|
||||
|
||||
def _build_thin_groups(client_type: str, app_config: AppConfig, client: ClientConfig, selected_source_names: list[str]) -> list[dict[str, Any]]:
|
||||
groups: list[dict[str, Any]] = []
|
||||
source_auto_names: list[str] = []
|
||||
|
||||
for source_name in selected_source_names:
|
||||
display_name = app_config.sources[source_name].display_name or source_name
|
||||
group_name = f"{display_name} 自动"
|
||||
source_auto_names.append(group_name)
|
||||
groups.append(
|
||||
{
|
||||
"name": group_name,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"use": [source_name],
|
||||
}
|
||||
)
|
||||
|
||||
if client_type == "mihomo":
|
||||
mixed_auto = {
|
||||
"name": client.mixed_auto_policy,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"include-all-providers": True,
|
||||
}
|
||||
manual = {
|
||||
"name": client.manual_policy,
|
||||
"type": "select",
|
||||
"proxies": [client.direct_policy],
|
||||
"include-all-providers": True,
|
||||
}
|
||||
else:
|
||||
mixed_auto = {
|
||||
"name": client.mixed_auto_policy,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"include-all": True,
|
||||
}
|
||||
manual = {
|
||||
"name": client.manual_policy,
|
||||
"type": "select",
|
||||
"proxies": [client.direct_policy],
|
||||
"include-all": True,
|
||||
}
|
||||
|
||||
groups.append(mixed_auto)
|
||||
|
||||
region_names = [region.name for region in app_config.regions.values()]
|
||||
selector_names = [*region_names, *[selector.name for selector in app_config.selector_groups]]
|
||||
|
||||
groups.append(
|
||||
{
|
||||
"name": client.source_policy,
|
||||
"type": "select",
|
||||
"proxies": [client.mixed_auto_policy, *source_auto_names, client.direct_policy],
|
||||
}
|
||||
)
|
||||
groups.append(manual)
|
||||
groups.append(
|
||||
{
|
||||
"name": client.main_policy,
|
||||
"type": "select",
|
||||
"proxies": [
|
||||
client.source_policy,
|
||||
client.mixed_auto_policy,
|
||||
*selector_names,
|
||||
client.manual_policy,
|
||||
client.direct_policy,
|
||||
],
|
||||
}
|
||||
)
|
||||
groups.extend(
|
||||
_build_custom_policy_groups(
|
||||
app_config=app_config,
|
||||
client=client,
|
||||
source_auto_names=source_auto_names,
|
||||
selector_names=selector_names,
|
||||
)
|
||||
)
|
||||
for region in app_config.regions.values():
|
||||
group = {
|
||||
"name": region.name,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"filter": region.filter,
|
||||
"tolerance": region.tolerance,
|
||||
}
|
||||
if client_type == "mihomo":
|
||||
group["include-all-providers"] = True
|
||||
else:
|
||||
group["include-all"] = True
|
||||
groups.append(group)
|
||||
|
||||
for selector in app_config.selector_groups:
|
||||
if selector.filter:
|
||||
groups.append(
|
||||
_build_thin_filter_group(
|
||||
client_type=client_type,
|
||||
client=client,
|
||||
group=selector,
|
||||
selected_source_names=selected_source_names,
|
||||
)
|
||||
)
|
||||
else:
|
||||
groups.append(
|
||||
{
|
||||
"name": selector.name,
|
||||
"type": selector.type,
|
||||
"proxies": _expand_proxy_tokens(
|
||||
selector.proxies,
|
||||
client=client,
|
||||
source_auto_names=source_auto_names,
|
||||
selector_names=selector_names,
|
||||
),
|
||||
}
|
||||
)
|
||||
return groups
|
||||
|
||||
|
||||
def _build_bundle_groups(
|
||||
app_config: AppConfig,
|
||||
client: ClientConfig,
|
||||
snapshots: list[SourceSnapshot],
|
||||
source_proxy_names: dict[str, list[str]],
|
||||
) -> list[dict[str, Any]]:
|
||||
groups: list[dict[str, Any]] = []
|
||||
source_auto_names: list[str] = []
|
||||
all_proxy_names = [name for names in source_proxy_names.values() for name in names]
|
||||
|
||||
for snapshot in snapshots:
|
||||
group_name = f"{snapshot.display_name} 自动"
|
||||
source_auto_names.append(group_name)
|
||||
groups.append(
|
||||
{
|
||||
"name": group_name,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"proxies": source_proxy_names.get(snapshot.name) or [client.direct_policy],
|
||||
}
|
||||
)
|
||||
|
||||
groups.append(
|
||||
{
|
||||
"name": client.mixed_auto_policy,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"proxies": all_proxy_names or [client.direct_policy],
|
||||
}
|
||||
)
|
||||
|
||||
region_names = [region.name for region in app_config.regions.values()]
|
||||
selector_names = [*region_names, *[selector.name for selector in app_config.selector_groups]]
|
||||
|
||||
groups.append(
|
||||
{
|
||||
"name": client.source_policy,
|
||||
"type": "select",
|
||||
"proxies": [client.mixed_auto_policy, *source_auto_names, client.direct_policy],
|
||||
}
|
||||
)
|
||||
groups.append(
|
||||
{
|
||||
"name": client.manual_policy,
|
||||
"type": "select",
|
||||
"proxies": [*all_proxy_names, client.direct_policy] if all_proxy_names else [client.direct_policy],
|
||||
}
|
||||
)
|
||||
groups.append(
|
||||
{
|
||||
"name": client.main_policy,
|
||||
"type": "select",
|
||||
"proxies": [
|
||||
client.source_policy,
|
||||
client.mixed_auto_policy,
|
||||
*selector_names,
|
||||
client.manual_policy,
|
||||
client.direct_policy,
|
||||
],
|
||||
}
|
||||
)
|
||||
groups.extend(
|
||||
_build_custom_policy_groups(
|
||||
app_config=app_config,
|
||||
client=client,
|
||||
source_auto_names=source_auto_names,
|
||||
selector_names=selector_names,
|
||||
)
|
||||
)
|
||||
for region in app_config.regions.values():
|
||||
matched = [name for name in all_proxy_names if re.search(region.filter, name)]
|
||||
groups.append(
|
||||
{
|
||||
"name": region.name,
|
||||
"type": "url-test",
|
||||
"url": str(client.test_url),
|
||||
"interval": client.test_interval,
|
||||
"tolerance": region.tolerance,
|
||||
"proxies": matched or [client.direct_policy],
|
||||
}
|
||||
)
|
||||
|
||||
for selector in app_config.selector_groups:
|
||||
if selector.filter:
|
||||
groups.append(
|
||||
_build_bundle_filter_group(
|
||||
client=client,
|
||||
group=selector,
|
||||
all_proxy_names=all_proxy_names,
|
||||
)
|
||||
)
|
||||
else:
|
||||
groups.append(
|
||||
{
|
||||
"name": selector.name,
|
||||
"type": selector.type,
|
||||
"proxies": _expand_proxy_tokens(
|
||||
selector.proxies,
|
||||
client=client,
|
||||
source_auto_names=source_auto_names,
|
||||
selector_names=selector_names,
|
||||
),
|
||||
}
|
||||
)
|
||||
return groups
|
||||
|
||||
@@ -1,73 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from app.models import ProxyNode, SourceSnapshot
|
||||
|
||||
|
||||
def source_auto_group_name(display_name: str) -> str:
|
||||
return f"{display_name} 自动"
|
||||
|
||||
|
||||
def proxy_dict_to_node(proxy: dict[str, Any]) -> ProxyNode | None:
|
||||
name = str(proxy.get("name", "")).strip()
|
||||
proxy_type = str(proxy.get("type", "")).strip()
|
||||
if not name or not proxy_type:
|
||||
return None
|
||||
|
||||
attrs = dict(proxy)
|
||||
attrs.pop("name", None)
|
||||
attrs.pop("type", None)
|
||||
server = attrs.pop("server", None)
|
||||
port = attrs.pop("port", None)
|
||||
udp = bool(attrs.pop("udp", True))
|
||||
return ProxyNode(
|
||||
name=name,
|
||||
type=proxy_type,
|
||||
server=server,
|
||||
port=port,
|
||||
udp=udp,
|
||||
attrs=attrs,
|
||||
)
|
||||
|
||||
|
||||
def dedupe_proxy_nodes(nodes: list[ProxyNode]) -> list[ProxyNode]:
|
||||
seen: set[str] = set()
|
||||
deduped: list[ProxyNode] = []
|
||||
for node in nodes:
|
||||
original = node.name
|
||||
name = original
|
||||
index = 2
|
||||
while name in seen:
|
||||
name = f"{original} #{index}"
|
||||
index += 1
|
||||
seen.add(name)
|
||||
deduped.append(node.model_copy(update={"name": name}))
|
||||
return deduped
|
||||
|
||||
|
||||
def build_bundle_proxy_inventory(
|
||||
snapshots: list[SourceSnapshot],
|
||||
) -> tuple[list[ProxyNode], dict[str, list[str]]]:
|
||||
all_nodes: list[ProxyNode] = []
|
||||
source_nodes: dict[str, list[ProxyNode]] = {}
|
||||
|
||||
for snapshot in snapshots:
|
||||
nodes: list[ProxyNode] = []
|
||||
for proxy in snapshot.document.proxies:
|
||||
node = proxy_dict_to_node(proxy)
|
||||
if node is not None:
|
||||
nodes.append(node)
|
||||
source_nodes[snapshot.name] = nodes
|
||||
all_nodes.extend(nodes)
|
||||
|
||||
deduped_all = dedupe_proxy_nodes(all_nodes)
|
||||
source_proxy_names: dict[str, list[str]] = {}
|
||||
cursor = 0
|
||||
for snapshot in snapshots:
|
||||
original_nodes = source_nodes[snapshot.name]
|
||||
count = len(original_nodes)
|
||||
source_proxy_names[snapshot.name] = [node.name for node in deduped_all[cursor : cursor + count]]
|
||||
cursor += count
|
||||
|
||||
return deduped_all, source_proxy_names
|
||||
@@ -1,122 +0,0 @@
|
||||
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
|
||||
@@ -1,69 +1,130 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
import re
|
||||
|
||||
import yaml
|
||||
|
||||
from app.config import get_settings
|
||||
from app.models import AppConfig, ClientConfig
|
||||
from app.services.profile_resolver import resolve_profile
|
||||
from app.services.rule_resolver import (
|
||||
build_rule_provider_entries as build_rule_provider_entries_for_profile,
|
||||
iter_resolved_rule_lines,
|
||||
load_rule_payload,
|
||||
load_rule_text,
|
||||
resolve_policy,
|
||||
)
|
||||
|
||||
|
||||
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)
|
||||
)
|
||||
|
||||
|
||||
_CIDR_PATTERN = re.compile(r"^[0-9a-fA-F:.]+/\d+$")
|
||||
|
||||
|
||||
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)
|
||||
line = ",".join(parts)
|
||||
else:
|
||||
line = f"{rendered_line},{target}"
|
||||
if append_no_resolve:
|
||||
line += ",no-resolve"
|
||||
return line
|
||||
|
||||
|
||||
def _resolve_rule_lines(rule_name: str, app_config: AppConfig, client: ClientConfig) -> list[str]:
|
||||
rule = app_config.rules[rule_name]
|
||||
target = resolve_policy(rule.policy, client)
|
||||
lines: list[str] = []
|
||||
|
||||
for payload_line in rule.payload:
|
||||
rendered = _render_payload_line(payload_line, rule.behavior)
|
||||
lines.append(_attach_policy(rendered, target, rule.no_resolve))
|
||||
|
||||
if rule.file:
|
||||
line = f"RULE-SET,{rule_name},{target}"
|
||||
if rule.no_resolve:
|
||||
line += ",no-resolve"
|
||||
lines.append(line)
|
||||
|
||||
return lines
|
||||
|
||||
|
||||
def build_rule_provider_entries(app_config: AppConfig, client: ClientConfig, base_url: str, public_path: str):
|
||||
resolved_profile = resolve_profile(
|
||||
app_config=app_config,
|
||||
client_type=_find_client_type(app_config, client),
|
||||
selected_source_names=list(app_config.sources.keys()),
|
||||
)
|
||||
return build_rule_provider_entries_for_profile(
|
||||
resolved_profile=resolved_profile,
|
||||
base_url=base_url,
|
||||
public_path=public_path,
|
||||
)
|
||||
providers: dict[str, dict] = {}
|
||||
for name, rule in app_config.rules.items():
|
||||
if not rule.file:
|
||||
continue
|
||||
entry = {
|
||||
"behavior": rule.behavior,
|
||||
"format": rule.format,
|
||||
"url": f"{base_url}/{public_path}/rules/{name}.yaml",
|
||||
"interval": rule.interval,
|
||||
}
|
||||
providers[name] = entry
|
||||
return providers
|
||||
|
||||
|
||||
def build_rule_set_references(app_config: AppConfig, client: ClientConfig) -> list[str]:
|
||||
resolved_profile = resolve_profile(
|
||||
app_config=app_config,
|
||||
client_type=_find_client_type(app_config, client),
|
||||
selected_source_names=list(app_config.sources.keys()),
|
||||
)
|
||||
return iter_resolved_rule_lines(
|
||||
resolved_profile=resolved_profile,
|
||||
include_rule_set_references=True,
|
||||
inline_file_payloads=False,
|
||||
)
|
||||
refs: list[str] = []
|
||||
for name in app_config.rules:
|
||||
refs.extend(_resolve_rule_lines(name, app_config, client))
|
||||
refs.append(f"MATCH,{client.main_policy}")
|
||||
return refs
|
||||
|
||||
|
||||
def build_inline_rules(app_config: AppConfig, client: ClientConfig) -> list[str]:
|
||||
resolved_profile = resolve_profile(
|
||||
app_config=app_config,
|
||||
client_type=_find_client_type(app_config, client),
|
||||
selected_source_names=list(app_config.sources.keys()),
|
||||
)
|
||||
return iter_resolved_rule_lines(
|
||||
resolved_profile=resolved_profile,
|
||||
include_rule_set_references=False,
|
||||
inline_file_payloads=True,
|
||||
)
|
||||
|
||||
|
||||
def _find_client_type(app_config: AppConfig, client: ClientConfig) -> str:
|
||||
for client_type, candidate in app_config.clients.items():
|
||||
if candidate == client:
|
||||
return client_type
|
||||
raise KeyError("client config not found in app config")
|
||||
|
||||
|
||||
__all__ = [
|
||||
"build_inline_rules",
|
||||
"build_rule_provider_entries",
|
||||
"build_rule_set_references",
|
||||
"load_rule_payload",
|
||||
"load_rule_text",
|
||||
"resolve_policy",
|
||||
]
|
||||
settings = get_settings()
|
||||
lines: list[str] = []
|
||||
for name, rule in app_config.rules.items():
|
||||
target = resolve_policy(rule.policy, client)
|
||||
for payload_line in rule.payload:
|
||||
rendered = _render_payload_line(payload_line, rule.behavior)
|
||||
lines.append(_attach_policy(rendered, target, rule.no_resolve))
|
||||
if not rule.file:
|
||||
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):
|
||||
rendered = _render_payload_line(payload_line, rule.behavior)
|
||||
lines.append(_attach_policy(rendered, target, rule.no_resolve))
|
||||
lines.append(f"MATCH,{client.main_policy}")
|
||||
return lines
|
||||
|
||||
Reference in New Issue
Block a user