fix(npu): expose advisory gateway on docker bridge

This commit is contained in:
William Valentin
2026-06-04 16:19:22 -07:00
parent 59c5fd3e57
commit aeb3c9f8fb
5 changed files with 57 additions and 14 deletions
+26 -2
View File
@@ -10,6 +10,7 @@ from __future__ import annotations
import argparse
import hashlib
import ipaddress
import json
import os
import sqlite3
@@ -21,6 +22,7 @@ from typing import Any, Callable
from urllib.parse import urlparse
HOST = "127.0.0.1"
DOCKER_BRIDGE_HOST = "172.19.0.1"
PORT = 18830
CLASSIFIER_URL = "http://127.0.0.1:18819/v1/classify"
GENAI_URL = "http://127.0.0.1:18820/v1/worker/generate"
@@ -40,6 +42,20 @@ AUTHORITY = {
}
def validate_bind_host(host: str, *, allow_docker_bridge: bool = False) -> None:
"""Restrict service exposure to localhost or the explicitly approved Docker bridge bind."""
if host == "127.0.0.1":
return
if not allow_docker_bridge:
raise ValueError("refusing non-local bind without --allow-docker-bridge")
try:
addr = ipaddress.ip_address(host)
except ValueError as exc:
raise ValueError("bind host must be a literal IP address") from exc
if host != DOCKER_BRIDGE_HOST or not (addr.version == 4 and addr.is_private and not addr.is_loopback and not addr.is_unspecified):
raise ValueError(f"Docker bridge bind must use approved bridge IP {DOCKER_BRIDGE_HOST}")
def sha256_text(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
@@ -335,9 +351,17 @@ def main(argv: list[str] | None = None) -> int:
parser.add_argument("--port", type=int, default=int(os.environ.get("NPU_ADVISORY_PORT", str(PORT))))
parser.add_argument("--log-db", default=str(DEFAULT_LOG_DB))
parser.add_argument("--allowed-root", action="append", dest="allowed_roots", default=None, help="Configured file root allowed for advisory doc/image triage. May be repeated.")
parser.add_argument(
"--allow-docker-bridge",
action="store_true",
default=os.environ.get("NPU_ADVISORY_ALLOW_DOCKER_BRIDGE", "").lower() in {"1", "true", "yes"},
help="Permit binding to a private Docker bridge IP instead of 127.0.0.1.",
)
args = parser.parse_args(argv)
if args.host != "127.0.0.1":
raise SystemExit("refusing non-local bind")
try:
validate_bind_host(args.host, allow_docker_bridge=args.allow_docker_bridge)
except ValueError as exc:
raise SystemExit(str(exc)) from exc
configured_roots = [Path(p).expanduser().resolve() for p in (args.allowed_roots or DEFAULT_ALLOWED_ROOTS)]
logger = AdvisoryLogger(args.log_db)
server = ThreadingHTTPServer((args.host, args.port), make_handler(logger, configured_roots))