Adopt nanoDNS-style INI config for the guide gateway

Replace the gateway's env-var DNS policy with docker/nanodns.ini, modelled on
https://github.com/drakmor/nanoDNS#config: [general] (redirect_url,
allowed_clients), [upstream] (ordered resolver list + timeout_ms), [overrides]
(mask = IPv4 spoof) and [exceptions] (masks forwarded upstream). Masks use
shell-style wildcards and match case-insensitively; resolution order is
exceptions -> overrides -> NXDOMAIN.

The unmatched-name behaviour is deliberately inverted from upstream nanoDNS:
names matching no rule return NXDOMAIN (default-deny) instead of being forwarded,
so PSN, updates and telemetry stay blocked. The source-client allowlist now lives
in [general] allowed_clients. The INI is baked into the image; CONFIG_FILE can
point at a mounted file for per-host overrides.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
2026-10-11 13:38:29 +01:00
co-authored by Claude Opus 4.8
parent 9f28b8e2a1
commit 50dd2d8266
7 changed files with 237 additions and 72 deletions
+123 -34
View File
@@ -1,3 +1,5 @@
import configparser
import fnmatch
import http.server
import ipaddress
import os
@@ -9,38 +11,96 @@ import struct
import threading
CONFIG_FILE = os.environ.get("CONFIG_FILE", "/srv/nanodns.ini")
DNS_PORT = int(os.environ.get("DNS_PORT", "8053"))
HTTP_PORT = int(os.environ.get("HTTP_PORT", "8081"))
HTTPS_PORT = int(os.environ.get("HTTPS_PORT", "8443"))
GATEWAY_IP = ipaddress.IPv4Address(os.environ.get("GATEWAY_IP", "192.168.20.30"))
REDIRECT_URL = os.environ.get("REDIRECT_URL", f"http://{GATEWAY_IP}:8000/")
TARGET_DOMAIN = "manuals.playstation.net"
# Exact names forwarded to the upstream resolver (homebrew store: catalog + downloads).
FORWARDED_DOMAINS = frozenset(
name.strip().lower()
for name in os.environ.get(
"FORWARDED_DOMAINS",
"homebrew.page,itsplk.github.io,store.spectrumlibrary.online,api.github.com,github.com,raw.githubusercontent.com,release-assets.githubusercontent.com,prosperopatches.com,cdn.prosperopatches.com,image.api.playstation.com,gmedia.playstation.com",
).split(",")
if name.strip()
)
UPSTREAM_DNS = str(ipaddress.ip_address(os.environ.get("UPSTREAM_DNS", "1.1.1.1")))
UPSTREAM_TIMEOUT = 3.0
# Source networks allowed to query the DNS service. Defaults to the console VLAN
# plus the server and trusted-management VLANs. Accepts a comma-separated mix of
# individual addresses and CIDR ranges; empty disables the filter and serves
# every client.
ALLOWED_CLIENTS = tuple(
ipaddress.ip_network(entry.strip(), strict=False)
for entry in os.environ.get(
"ALLOWED_CLIENTS",
"192.168.40.0/28,192.168.20.0/24,192.168.10.0/27",
).split(",")
if entry.strip()
)
CERTIFICATE = "/srv/certificates/manuals.crt"
PRIVATE_KEY = "/srv/certificates/manuals.key"
# Built-in defaults, used when the config file (or a given key) is absent. They
# mirror the reviewed production policy: the User Guide domain is spoofed to the
# local gateway, the reviewed store/update hosts are forwarded upstream, and
# every other name returns NXDOMAIN (default-deny).
DEFAULT_REDIRECT_URL = "http://192.168.20.30:8000/"
DEFAULT_ALLOWED_CLIENTS = "192.168.40.0/28, 192.168.20.0/24, 192.168.10.0/27"
DEFAULT_UPSTREAMS = "1.1.1.1"
DEFAULT_TIMEOUT_MS = "3000"
DEFAULT_OVERRIDES = (("manuals.playstation.net", "192.168.20.30"),)
DEFAULT_EXCEPTIONS = (
"homebrew.page",
"itsplk.github.io",
"store.spectrumlibrary.online",
"api.github.com",
"github.com",
"raw.githubusercontent.com",
"release-assets.githubusercontent.com",
"prosperopatches.com",
"cdn.prosperopatches.com",
"image.api.playstation.com",
"gmedia.playstation.com",
)
def _split(value):
return [item for item in value.replace(",", " ").split() if item]
def load_config(path):
parser = configparser.ConfigParser(
allow_no_value=True,
inline_comment_prefixes=("#", ";"),
)
# Keep mask/address case verbatim; query names are lowercased before lookup.
parser.optionxform = str
parser.read(path)
general = parser["general"] if parser.has_section("general") else {}
redirect_url = general.get("redirect_url") or DEFAULT_REDIRECT_URL
allowed_raw = general.get("allowed_clients") or DEFAULT_ALLOWED_CLIENTS
allowed_clients = tuple(
ipaddress.ip_network(entry, strict=False) for entry in _split(allowed_raw)
)
upstream = parser["upstream"] if parser.has_section("upstream") else {}
servers_raw = (
upstream.get("server") or upstream.get("servers") or DEFAULT_UPSTREAMS
)
upstreams = tuple(
str(ipaddress.ip_address(server)) for server in _split(servers_raw)
)
timeout = max(0.1, int(upstream.get("timeout_ms") or DEFAULT_TIMEOUT_MS) / 1000.0)
override_items = (
parser.items("overrides")
if parser.has_section("overrides")
else DEFAULT_OVERRIDES
)
overrides = tuple(
(mask.lower(), ipaddress.IPv4Address(address))
for mask, address in override_items
if address is not None
)
exception_items = (
parser.items("exceptions")
if parser.has_section("exceptions")
else ((mask, None) for mask in DEFAULT_EXCEPTIONS)
)
exceptions = tuple(mask.lower() for mask, _ in exception_items)
return redirect_url, allowed_clients, upstreams, timeout, overrides, exceptions
(
REDIRECT_URL,
ALLOWED_CLIENTS,
UPSTREAMS,
UPSTREAM_TIMEOUT,
OVERRIDES,
EXCEPTIONS,
) = load_config(CONFIG_FILE)
def client_allowed(address):
if not ALLOWED_CLIENTS:
@@ -52,6 +112,17 @@ def client_allowed(address):
return any(client in network for network in ALLOWED_CLIENTS)
def match_any(name, masks):
return any(fnmatch.fnmatchcase(name, mask) for mask in masks)
def override_for(name):
for mask, address in OVERRIDES:
if fnmatch.fnmatchcase(name, mask):
return address
return None
def parse_question(packet):
if len(packet) < 17:
raise ValueError("short DNS query")
@@ -104,12 +175,12 @@ def receive_exactly(connection, wanted):
return b"".join(chunks)
def forward_query(packet, use_tcp=False):
family = socket.AF_INET6 if ":" in UPSTREAM_DNS else socket.AF_INET
def _forward_one(packet, server, use_tcp):
family = socket.AF_INET6 if ":" in server else socket.AF_INET
socket_type = socket.SOCK_STREAM if use_tcp else socket.SOCK_DGRAM
with socket.socket(family, socket_type) as upstream:
upstream.settimeout(UPSTREAM_TIMEOUT)
upstream.connect((UPSTREAM_DNS, 53))
upstream.connect((server, 53))
if use_tcp:
upstream.sendall(struct.pack("!H", len(packet)) + packet)
response_length = struct.unpack("!H", receive_exactly(upstream, 2))[0]
@@ -123,6 +194,16 @@ def forward_query(packet, use_tcp=False):
return response
def forward_query(packet, use_tcp=False):
last_error = OSError("no upstream resolver configured")
for server in UPSTREAMS:
try:
return _forward_one(packet, server, use_tcp)
except (OSError, ValueError) as error:
last_error = error
raise last_error
def dns_response(packet, use_tcp=False):
if len(packet) < 12:
return b""
@@ -137,21 +218,25 @@ def dns_response(packet, use_tcp=False):
return struct.pack("!HHHHHH", transaction_id, response_flags | 1, 0, 0, 0, 0)
question = packet[12:question_end]
if name in FORWARDED_DOMAINS:
# Exceptions bypass overrides and are forwarded upstream (nanoDNS semantics).
if match_any(name, EXCEPTIONS):
try:
return forward_query(packet, use_tcp=use_tcp)
except (OSError, ValueError) as error:
print(f"DNS upstream failure for {name}: {error}", flush=True)
return error_response(packet, 2) # SERVFAIL
if name != TARGET_DOMAIN:
address = override_for(name)
if address is None:
# Default-deny: a name matching no rule returns NXDOMAIN.
return struct.pack("!HHHHHH", transaction_id, response_flags | 3, 1, 0, 0, 0) + question
if qclass == 1 and qtype in (1, 255):
answer = (
b"\xc0\x0c"
+ struct.pack("!HHIH", 1, 1, 60, 4)
+ GATEWAY_IP.packed
+ address.packed
)
return struct.pack("!HHHHHH", transaction_id, response_flags, 1, 1, 0, 0) + question + answer
@@ -274,9 +359,13 @@ if __name__ == "__main__":
serve(http, "http-redirect")
serve(https, "https-redirect")
overrides_summary = (
", ".join(f"{mask}->{address}" for mask, address in OVERRIDES) or "none"
)
print(
f"PS5 guide gateway: {TARGET_DOMAIN} -> {GATEWAY_IP}; redirect -> {REDIRECT_URL}; "
f"forward {','.join(sorted(FORWARDED_DOMAINS))} -> {UPSTREAM_DNS}; "
"PS5 guide gateway (nanoDNS-style, default-deny): "
f"redirect -> {REDIRECT_URL}; overrides {overrides_summary}; "
f"forward {','.join(EXCEPTIONS) or 'none'} -> {','.join(UPSTREAMS) or 'none'}; "
f"clients {','.join(str(network) for network in ALLOWED_CLIENTS) or 'any'}",
flush=True,
)