Security + upstream fixes
- Network core moves to a root-only folder; BroV talks to a narrow root helper (netctl.py) over a user-only socket; install script verifies the mihomo SHA256 and migrates keys (scripts/netcore) - Hardened runtime, no get-task-allow; bypassPermissions removed from the chat; concealed clipboard items are not restored; DangerCheck knows core, LaunchAgents and hook paths; dropped-file copies expire after 7 days - Ported from upstream Coucou: 1h crash fix (d05f22b), safe settings.json writes (918d30e), Escape/fold for pending approvals (6012900, 40e3ba8), auto-close delay + reopen (74984f2, ea244a7), full AskUserQuestion (52b1562) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
maksarsanjeev
Executable
+226
@@ -0,0 +1,226 @@
|
||||
#!/usr/bin/env python3
|
||||
"""BroV network core prototype: builds core/config.yaml for mihomo from src/*.
|
||||
|
||||
keys/*.vpnkey one vpn:// key per Amnezia connection (AmneziaWG 2/3); BroV adds them
|
||||
and rewrites keys/amnezia.yaml itself, gen.py does the same on a full build
|
||||
src/saga.conf WireGuard client config -> only 192.168.8.0/24
|
||||
src/planet9.conf WireGuard client config -> only 192.168.68.0/24
|
||||
src/vless.sub 3x-ui subscription URL (all VLESS nodes)
|
||||
|
||||
Secrets stay in this private folder; nothing here goes to git.
|
||||
"""
|
||||
import base64, json, os, re, secrets, zlib, configparser
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
SRC = os.path.join(HERE, "src")
|
||||
KEYS = os.path.join(HERE, "keys")
|
||||
# Installed next to its inputs in /Library/Application Support/BroV/netcore (root, 0700):
|
||||
# the root core reads its config from here, nothing user-writable is involved.
|
||||
CORE = HERE
|
||||
|
||||
HOME_NET = "192.168.10.0/24"
|
||||
CLIENTS = { # name: (conf file, routed subnets)
|
||||
"saga": ("saga.conf", ["192.168.8.0/24", "10.0.0.0/24"]),
|
||||
"planet9": ("planet9.conf", ["192.168.68.0/24", "172.3.3.0/24"]),
|
||||
}
|
||||
AI_DOMAINS = ["anthropic.com", "claude.ai", "claude.com", "openai.com", "chatgpt.com",
|
||||
"oaistatic.com", "oaiusercontent.com", "github.com", "githubusercontent.com"]
|
||||
|
||||
|
||||
def mid(v, default):
|
||||
"""'100-120' -> 110 (mihomo takes single ints for timers)."""
|
||||
if v is None or v == "":
|
||||
return default
|
||||
m = re.match(r"^\s*(\d+)\s*-\s*(\d+)\s*$", str(v))
|
||||
return (int(m.group(1)) + int(m.group(2))) // 2 if m else int(v)
|
||||
|
||||
|
||||
def amnezia(path, name):
|
||||
key = open(path).read().strip()[len("vpn://"):]
|
||||
key += "=" * (-len(key) % 4)
|
||||
j = json.loads(zlib.decompress(base64.urlsafe_b64decode(key)[4:]))
|
||||
awg = j["containers"][0]["awg"]
|
||||
c = awg["last_config"] if isinstance(awg["last_config"], dict) else json.loads(awg["last_config"])
|
||||
ver = 3 if c.get("HeaderProtectionKey") else 2
|
||||
opt = {"version": ver, "jc": int(c["Jc"]), "jmin": int(c["Jmin"]), "jmax": int(c["Jmax"]),
|
||||
"s1": int(c["S1"]), "s2": int(c["S2"])}
|
||||
for k in ("S3", "S4"):
|
||||
if c.get(k):
|
||||
opt[k.lower()] = int(c[k])
|
||||
for k in ("H1", "H2", "H3", "H4"):
|
||||
v = c[k]
|
||||
opt[k.lower()] = int(v) if str(v).isdigit() else v
|
||||
for k in ("I1", "I2", "I3", "I4", "I5"):
|
||||
if c.get(k):
|
||||
opt[k.lower()] = c[k]
|
||||
if ver == 3:
|
||||
opt.update({
|
||||
"header-protection-key": c["HeaderProtectionKey"],
|
||||
"content-padding-addition": c.get("ContentPaddingAddition", "0"),
|
||||
"rekey-after-time": mid(c.get("RekeyAfterTime"), 120),
|
||||
"rekey-timeout": mid(c.get("RekeyTimeout"), 5),
|
||||
"reject-after-time": mid(c.get("RejectAfterTime"), 180),
|
||||
"keepalive-timeout": mid(c.get("KeepaliveTimeout"), 10),
|
||||
"max-handshake-attempts": mid(c.get("MaxHandshakeAttempts"), 18),
|
||||
"random-trailers": c.get("RandomTrailers") == "on",
|
||||
"disable-cookies": c.get("DisableCookies") == "on",
|
||||
})
|
||||
return {
|
||||
"name": name, "type": "wireguard", "server": c["hostName"], "port": int(c["port"]),
|
||||
"ip": c["client_ip"], "private-key": c["client_priv_key"], "public-key": c["server_pub_key"],
|
||||
"pre-shared-key": c.get("psk_key") or None, "mtu": int(c.get("mtu", 1376)), "udp": True,
|
||||
"persistent-keepalive": mid(c.get("persistent_keep_alive"), 25),
|
||||
"amnezia-wg-option": opt,
|
||||
}
|
||||
|
||||
|
||||
def wg(name, path):
|
||||
p = configparser.ConfigParser()
|
||||
p.optionxform = str
|
||||
p.read(os.path.join(SRC, path))
|
||||
i, peer = p["Interface"], p["Peer"]
|
||||
host, port = peer["Endpoint"].rsplit(":", 1)
|
||||
out = {"name": name, "type": "wireguard", "server": host, "port": int(port),
|
||||
"ip": i["Address"].split("/")[0], "private-key": i["PrivateKey"],
|
||||
"public-key": peer["PublicKey"], "mtu": int(i.get("MTU", 1420)), "udp": True}
|
||||
if peer.get("PresharedKey"):
|
||||
out["pre-shared-key"] = peer["PresharedKey"]
|
||||
if peer.get("PersistentKeepalive"):
|
||||
out["persistent-keepalive"] = int(peer["PersistentKeepalive"])
|
||||
return out
|
||||
|
||||
|
||||
def y(v, ind=0):
|
||||
"""Tiny YAML emitter (no PyYAML dependency)."""
|
||||
pad = " " * ind
|
||||
if isinstance(v, dict):
|
||||
lines = []
|
||||
for k, x in v.items():
|
||||
if x is None:
|
||||
continue
|
||||
if isinstance(x, (dict, list)) and x:
|
||||
lines.append(f"{pad}{k}:\n{y(x, ind + 1)}")
|
||||
else:
|
||||
lines.append(f"{pad}{k}: {scalar(x)}")
|
||||
return "\n".join(lines)
|
||||
if isinstance(v, list):
|
||||
lines = []
|
||||
for x in v:
|
||||
if isinstance(x, dict):
|
||||
body = y(x, ind + 1).lstrip()
|
||||
lines.append(f"{pad}- {body}")
|
||||
else:
|
||||
lines.append(f"{pad}- {scalar(x)}")
|
||||
return "\n".join(lines)
|
||||
return pad + scalar(v)
|
||||
|
||||
|
||||
def scalar(x):
|
||||
if isinstance(x, bool):
|
||||
return "true" if x else "false"
|
||||
if isinstance(x, (int, float)):
|
||||
return str(x)
|
||||
if isinstance(x, list) and not x:
|
||||
return "[]"
|
||||
return json.dumps(str(x), ensure_ascii=False)
|
||||
|
||||
|
||||
def main():
|
||||
os.makedirs(CORE, exist_ok=True)
|
||||
secret_file = os.path.join(CORE, "api.secret")
|
||||
if not os.path.exists(secret_file):
|
||||
fd = os.open(secret_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
||||
with os.fdopen(fd, "w") as f:
|
||||
f.write(secrets.token_urlsafe(24))
|
||||
api_secret = open(secret_file).read().strip()
|
||||
|
||||
# Amnezia connections live in their own provider file so BroV can add keys live.
|
||||
keyfiles = sorted(f for f in os.listdir(KEYS) if f.endswith(".vpnkey"))
|
||||
awg = [amnezia(os.path.join(KEYS, f), "AWG " + f[:-len(".vpnkey")]) for f in keyfiles]
|
||||
prov = os.path.join(KEYS, "amnezia.yaml")
|
||||
with open(os.open(prov, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "w") as f:
|
||||
f.write(y({"proxies": awg}) + "\n")
|
||||
proxies = [wg(n, f) for n, (f, _) in CLIENTS.items()]
|
||||
sub = open(os.path.join(SRC, "vless.sub")).read().strip()
|
||||
|
||||
rules = [f"IP-CIDR,{HOME_NET},DIRECT,no-resolve", "IP-CIDR,127.0.0.0/8,DIRECT,no-resolve",
|
||||
# Home's own public IP (RustDesk, Gitea): never via a foreign exit.
|
||||
"IP-CIDR,79.111.14.0/32,DIRECT,no-resolve", "DOMAIN-SUFFIX,sanjeev.ru,DIRECT"]
|
||||
for name, (_, nets) in CLIENTS.items():
|
||||
# Through a switch group: BroV turns client networks on/off without a reload.
|
||||
rules += [f"IP-CIDR,{n},{name}-sw,no-resolve" for n in nets]
|
||||
rules += [f"DOMAIN-SUFFIX,{d},ai-out" for d in AI_DOMAINS]
|
||||
rules += ["DOMAIN-SUFFIX,ru,DIRECT", "DOMAIN-SUFFIX,su,DIRECT", "DOMAIN-SUFFIX,xn--p1ai,DIRECT",
|
||||
"MATCH,ai-out"]
|
||||
|
||||
cfg = {
|
||||
"mixed-port": 7890, "allow-lan": False, "mode": "rule", "log-level": "error", "ipv6": False,
|
||||
"external-controller": "127.0.0.1:9097", "secret": api_secret, "unified-delay": True,
|
||||
"find-process-mode": "strict",
|
||||
"profile": {"store-selected": True},
|
||||
"tun": {"enable": True, "stack": "mixed", "auto-route": True, "auto-detect-interface": True,
|
||||
"dns-hijack": ["any:53"], "mtu": 1400},
|
||||
"dns": {"enable": True, "ipv6": False, "enhanced-mode": "fake-ip", "fake-ip-range": "198.18.0.1/16",
|
||||
"fake-ip-filter": ["*.lan", "*.local", "+.duckdns.org"],
|
||||
"default-nameserver": ["77.88.8.8", "1.1.1.1"],
|
||||
"proxy-server-nameserver": ["77.88.8.8", "1.1.1.1"],
|
||||
"nameserver": ["https://1.1.1.1/dns-query#ai-out", "https://8.8.8.8/dns-query#ai-out"],
|
||||
"direct-nameserver": ["77.88.8.8", "77.88.8.1"]},
|
||||
"proxies": proxies,
|
||||
"proxy-providers": {
|
||||
"amnezia-keys": {"type": "file", "path": "./keys/amnezia.yaml",
|
||||
"health-check": {"enable": True, "url": "https://www.gstatic.com/generate_204",
|
||||
"interval": 300}},
|
||||
"vless-cluster": {
|
||||
# Fetch the list directly: the nodes themselves are dialled directly anyway.
|
||||
"type": "http", "url": sub, "interval": 43200, "path": "./providers/vless.yaml", "proxy": "DIRECT",
|
||||
"health-check": {"enable": True, "url": "https://www.gstatic.com/generate_204", "interval": 300}}},
|
||||
"proxy-groups": [
|
||||
# What BroV's globe panel switches: "auto", the Amnezia group, or one VLESS node.
|
||||
{"name": "ai-out", "type": "select", "proxies": ["auto", "amnezia"], "use": ["vless-cluster"]},
|
||||
# Fastest alive exit among every Amnezia connection and every VLESS node;
|
||||
# switches only when another one is 100+ ms faster.
|
||||
{"name": "auto", "type": "url-test", "use": ["amnezia-keys", "vless-cluster"],
|
||||
"url": "https://www.gstatic.com/generate_204", "interval": 120, "tolerance": 100, "lazy": False},
|
||||
# Amnezia: "amnezia-auto" (fastest connection) or one fixed connection.
|
||||
{"name": "amnezia", "type": "select", "proxies": ["amnezia-auto"], "use": ["amnezia-keys"]},
|
||||
{"name": "amnezia-auto", "type": "url-test", "use": ["amnezia-keys"],
|
||||
"url": "https://www.gstatic.com/generate_204", "interval": 120, "tolerance": 100, "lazy": False},
|
||||
] + [
|
||||
# Client networks: off (REJECT) until switched on in BroV.
|
||||
{"name": f"{n}-sw", "type": "select", "proxies": ["REJECT", n]} for n in CLIENTS
|
||||
],
|
||||
"rules": rules,
|
||||
}
|
||||
# The API secret never leaves this root-only folder: BroV goes through netctl.py.
|
||||
path = os.path.join(CORE, "config.yaml")
|
||||
with open(os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "w") as f:
|
||||
f.write("# Generated by gen.py — do not edit by hand, do not share.\n" + y(cfg) + "\n")
|
||||
print("wrote", path, "|", len(proxies), "proxies + vless subscription |", len(rules), "rules")
|
||||
|
||||
|
||||
def check(path):
|
||||
"""--check <file>: is this a usable Amnezia key? Prints JSON for BroV."""
|
||||
try:
|
||||
raw = open(path).read().strip()
|
||||
if not raw.startswith("vpn://"):
|
||||
raise ValueError("ключ должен начинаться с vpn://")
|
||||
key = raw[len("vpn://"):]
|
||||
key += "=" * (-len(key) % 4)
|
||||
j = json.loads(zlib.decompress(base64.urlsafe_b64decode(key)[4:]))
|
||||
cont = j["containers"][0]
|
||||
if "awg" not in cont:
|
||||
raise ValueError("это не AmneziaWG (контейнер %s) — пока поддерживается только AmneziaWG" % cont.get("container"))
|
||||
p = amnezia(path, "check")
|
||||
print(json.dumps({"ok": True, "name": j.get("description") or p["server"], "server": p["server"],
|
||||
"port": p["port"], "version": p["amnezia-wg-option"]["version"]}, ensure_ascii=False))
|
||||
except Exception as e:
|
||||
print(json.dumps({"ok": False, "error": str(e) or e.__class__.__name__}, ensure_ascii=False))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
if len(sys.argv) == 3 and sys.argv[1] == "--check":
|
||||
check(sys.argv[2])
|
||||
else:
|
||||
main()
|
||||
Reference in New Issue
Block a user