Restrict guide gateway DNS to the console VLAN
Add an ALLOWED_CLIENTS source-address filter to the DNS service so it only answers queries from the authorized PS5 VLAN (192.168.40.0/28 by default). Queries from any other source address are dropped and logged as "denied" on both UDP and TCP. The value accepts a comma-separated mix of individual addresses and CIDR ranges, and an empty value disables the filter. Record the two authorized consoles (PS5PRO 192.168.40.14, PS5FAT 192.168.40.13) and the stale 192.168.10.x workstation address in the docs. Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
+26
-1
@@ -26,10 +26,28 @@ FORWARDED_DOMAINS = frozenset(
|
||||
)
|
||||
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 (PS5 VLAN by default).
|
||||
# Accepts comma-separated 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").split(",")
|
||||
if entry.strip()
|
||||
)
|
||||
CERTIFICATE = "/srv/certificates/manuals.crt"
|
||||
PRIVATE_KEY = "/srv/certificates/manuals.key"
|
||||
|
||||
|
||||
def client_allowed(address):
|
||||
if not ALLOWED_CLIENTS:
|
||||
return True
|
||||
try:
|
||||
client = ipaddress.ip_address(address)
|
||||
except ValueError:
|
||||
return False
|
||||
return any(client in network for network in ALLOWED_CLIENTS)
|
||||
|
||||
|
||||
def parse_question(packet):
|
||||
if len(packet) < 17:
|
||||
raise ValueError("short DNS query")
|
||||
@@ -150,6 +168,9 @@ class ThreadingTCPServer(socketserver.ThreadingTCPServer):
|
||||
class UDPDNSHandler(socketserver.BaseRequestHandler):
|
||||
def handle(self):
|
||||
packet, connection = self.request
|
||||
if not client_allowed(self.client_address[0]):
|
||||
print(f"DNS UDP {self.client_address[0]} denied", flush=True)
|
||||
return
|
||||
try:
|
||||
name, qtype, _, _ = parse_question(packet)
|
||||
print(f"DNS UDP {self.client_address[0]} {name} type={qtype}", flush=True)
|
||||
@@ -162,6 +183,9 @@ class UDPDNSHandler(socketserver.BaseRequestHandler):
|
||||
|
||||
class TCPDNSHandler(socketserver.BaseRequestHandler):
|
||||
def handle(self):
|
||||
if not client_allowed(self.client_address[0]):
|
||||
print(f"DNS TCP {self.client_address[0]} denied", flush=True)
|
||||
return
|
||||
length_bytes = self._receive_exactly(2)
|
||||
if length_bytes is None:
|
||||
return
|
||||
@@ -248,7 +272,8 @@ if __name__ == "__main__":
|
||||
|
||||
print(
|
||||
f"PS5 guide gateway: {TARGET_DOMAIN} -> {GATEWAY_IP}; redirect -> {REDIRECT_URL}; "
|
||||
f"forward {','.join(sorted(FORWARDED_DOMAINS))} -> {UPSTREAM_DNS}",
|
||||
f"forward {','.join(sorted(FORWARDED_DOMAINS))} -> {UPSTREAM_DNS}; "
|
||||
f"clients {','.join(str(network) for network in ALLOWED_CLIENTS) or 'any'}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user