Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 90 additions & 24 deletions amplifier_web/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,10 @@

import argparse
import asyncio
from contextlib import contextmanager, ExitStack
import errno
import os
import socket
import sys
from pathlib import Path
import threading
Expand Down Expand Up @@ -233,33 +235,110 @@ def _serve(args, data_dir: Path, *, task_runner=None) -> None:
_serve_unlocked(args, data_dir, task_runner=task_runner)


@contextmanager
def _listener_sockets(binds, port):
"""Reserve real listeners before application startup, without a rebind gap.

Each configured name must succeed. Like asyncio's per-name resolution, an
unavailable address family may yield to another result of that same name.
Keep raw bind errors: asyncio discards EADDRNOTAVAIL and loses its errno.
"""
with ExitStack() as owned:
sockets = []
try:
for host in binds:
infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM,
flags=socket.AI_PASSIVE)
if not infos:
raise socket.gaierror(socket.EAI_NONAME, "No listener addresses")
available = []
unavailable = None
for family, kind, protocol, _, address in dict.fromkeys(infos):
try:
sock = socket.socket(family, kind, protocol)
except OSError as exc:
if exc.errno not in {errno.EAFNOSUPPORT, errno.EPROTONOSUPPORT}:
raise
unavailable = exc
continue
owned.callback(sock.close)
if os.name == "posix" and sys.platform != "cygwin":
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
if family == socket.AF_INET6 and hasattr(socket, "IPPROTO_IPV6"):
sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, True)
try:
sock.bind(address)
except OSError as exc:
if exc.errno != errno.EADDRNOTAVAIL:
raise
sock.close()
unavailable = exc
continue
# Strong reservation, including overlapping wildcard binds.
# TCP may queue here; authenticated HTTP health is not ready.
sock.listen(128)
sock.setblocking(False)
available.append(sock)
if not available:
assert unavailable is not None
raise unavailable
sockets.extend(available)
except socket.gaierror:
# EAI codes can collide numerically with ordinary errno values.
raise SystemExit(
f"Could not resolve configured bind address {host}, port {port}.\n"
"Check the bind hostname and this computer's name resolution. "
"No running process was stopped or replaced."
) from None
except OSError as exc:
guidance = {
errno.EADDRINUSE: 'The address or port is already in use. Check the existing service with amplifier-unified service status, or choose another port with --port.',
errno.EADDRNOTAVAIL: 'The configured bind address is not available on this computer. Check its current network addresses and your bind setting.',
errno.EACCES: 'Permission was denied. Check permission to use the configured address and port; choose an allowed port with --port if needed.',
errno.EAFNOSUPPORT: 'The configured address family is not supported on this computer. Check its network support and your bind setting.',
errno.EPROTONOSUPPORT: 'The configured address family is not supported on this computer. Check its network support and your bind setting.',
}.get(exc.errno)
if guidance is None:
raise
raise SystemExit(
f"Could not start Amplifier on {host}, port {port}.\n"
f"{guidance}\nNo running process was stopped or replaced."
) from None
# Do not classify exceptions from application construction or serving as
# bind failures. The same owned sockets close on every exit, even when
# aiohttp never attaches them or only attaches a prefix.
yield sockets


def _serve_unlocked(args, data_dir: Path, *, task_runner=None) -> None:
from aiohttp import web
from .deployment import load_server_config
overrides = _server_overrides(args)
config = load_server_config(data_dir, overrides=overrides)
from .tls import ssl_context
config = load_server_config(data_dir, overrides=_server_overrides(args))
os.environ["AMPLIFIER_WEB_HOME"] = str(data_dir)
secure = ssl_context(data_dir, config)
with _listener_sockets(config["bind"], config["port"]) as sockets:
_serve_listeners(args, data_dir, config, secure, sockets, task_runner=task_runner)


def _serve_listeners(args, data_dir, config, secure, sockets, *, task_runner=None):
from aiohttp import web
from .settings_migration import migrate_settings
migrate_settings(data_dir, [args.workspace])
from .host.config import load_config
load_config(args.workspace)
from .server import create_app
from .tls import ssl_context
secure = ssl_context(data_dir, config)
scheme = "https" if secure else "http"
browser_host = {'0.0.0.0': '127.0.0.1', '::': '::1'}.get(config['bind'][0], config['bind'][0])
if ':' in browser_host:
browser_host = f'[{browser_host}]'
url = (config["public_origins"][0] if config["public_origins"] else
f"{scheme}://{browser_host}:{config['port']}")
timer = None
listener_ready = False

def listening(_message):
# aiohttp calls this only after every configured socket has bound.
# Application startup signals run earlier and cannot establish this.
nonlocal timer, listener_ready
listener_ready = True
# aiohttp calls this only after every held socket has attached.
# TCP reservation and application startup alone cannot establish this.
nonlocal timer
if task_runner is not None:
task_runner.phase("listening")
print(f"Amplifier Unified · listening on {', '.join(config['bind'])}, port {config['port']} ({scheme})\n"
Expand Down Expand Up @@ -287,20 +366,7 @@ async def managed_app():
return app
app = managed_app()
web.run_app(app,
host=config["bind"], port=config["port"], ssl_context=secure, print=listening, access_log_class=SafeAccessLogger, access_log_format='%a %t "%r" %s %b')
except OSError as exc:
guidance = {
errno.EADDRINUSE: 'The address or port is already in use. Check the existing service with amplifier-unified service status, or choose another port with --port.',
errno.EADDRNOTAVAIL: 'The configured bind address is not available on this computer. Check its current network addresses and your bind setting.',
errno.EACCES: 'Permission was denied. Check permission to use the configured address and port; choose an allowed port with --port if needed.',
}.get(exc.errno)
# A filesystem error during application startup is not a bind failure.
if listener_ready or guidance is None or exc.filename is not None:
raise
raise SystemExit(
f"Could not start Amplifier on {', '.join(config['bind'])}, port {config['port']}.\n"
f"{guidance}\nNo running process was stopped or replaced."
) from None
sock=sockets, ssl_context=secure, print=listening, access_log_class=SafeAccessLogger, access_log_format='%a %t "%r" %s %b')
except PublishingError as exc:
# The same library error can also mean unsupported OS locking. Only an
# actual held-lock conflict establishes that another process owns it.
Expand Down
18 changes: 18 additions & 0 deletions docs/DEPLOYMENT.md
Original file line number Diff line number Diff line change
Expand Up @@ -242,6 +242,24 @@ Foreground `serve` announces the listener and schedules browser opening only
after all configured addresses bind successfully. An occupied port, unavailable
address or permission refusal gives a specific next step. It does not stop an
existing process, choose another interface, or change the saved configuration.
Name-resolution failures give hostname guidance rather than a port-conflict
diagnosis. A configured name may use its available address family, but success
for one configured entry never substitutes for another unavailable entry.

The CLI acquires and holds the actual listening sockets before settings migration
or application construction. A bind failure therefore starts no app controllers
and creates no new app diagnostic capture. Every held socket is released on a
failed start; successful startup transfers the same sockets to the HTTP server
without closing and rebinding. TCP connections may queue during construction:
a TCP connect alone is **not** readiness. Continue to use the authenticated
HTTP(S) health checks above. Later application/TLS/filesystem failures keep their
own identity rather than being reported as bind errors.

Service-manager retry policies are unchanged, including launchd's `KeepAlive`.
An unavailable VPN address still causes failed starts until it returns or the
operator deliberately changes the configuration or stops the service. This fix
does not suppress process retries, rewrite service definitions or repair/delete
historical failed-start captures.

If the generated definition differs, rerun the same install command with `--replace` to save a timestamped private backup and restart with the new definition. The command reports the backup location. Custom service files are preserved unless `--replace` is explicit; review them before replacing. Expected service setup errors are reported without a Python traceback and do not undo a separately completed package installation.

Expand Down
37 changes: 31 additions & 6 deletions tests/fixtures/windows_user_trial_probe.py
Original file line number Diff line number Diff line change
Expand Up @@ -1891,6 +1891,8 @@ def run(wheel, receipt):
CLI_FUNCTIONS = {
"cli_main": ("amplifier_web.cli", "main"),
"cli_serve": ("amplifier_web.cli", "_serve_unlocked"),
"cli_listeners": ("amplifier_web.cli", "_listener_sockets.__wrapped__"),
"cli_serve_listeners": ("amplifier_web.cli", "_serve_listeners"),
"app_dispatch": ("amplifier_web.service", "AppService.dispatch"),
"app_dispatch_action": ("amplifier_web.service", "AppService._dispatch"),
"app_preferences": ("amplifier_web.service", "AppService._refresh_shared_preferences"),
Expand Down Expand Up @@ -1978,24 +1980,47 @@ def cli_arguments(mode, root, port):


class CliNetworkGuard(NetworkGuard):
"""One real asyncio CLI listener, not a loopback/network permission window."""
"""One real configured CLI listener, not a loopback/network permission window."""
def __init__(self, log, port, *, serve=False, connect=False):
super().__init__(log)
self.port = port if connect else None
self.target_port, self.serve, self.listener = port, serve, None
self.resolved = False

def audit(self, event, args):
if event not in NETWORK_EVENTS:
return
# Only loaded public functions get names/fingerprints; resolve at the
# event because main may import a source after the guard was created.
self.codes.update(cli_codes())
if (event == "socket.bind" and self.serve and self.listener is None
and args[1] == ("127.0.0.1", self.target_port)):
if self.serve:
frame = sys._getframe(1)
# Actual CPython create_server binds this exact socket; no patched
# resolver/binder/run_app and no generic numeric/DNS exemption.
if (frame.f_code is asyncio.BaseEventLoop.create_server.__code__
cli = sys.modules.get("amplifier_web.cli")
function = getattr(getattr(cli, "_listener_sockets", None), "__wrapped__", None)
code = getattr(function, "__code__", None)
# The new production owner reserves sockets before app construction.
# Only this loaded helper may resolve its ONE configured numeric
# loopback entry, through the real stdlib wrapper with exact options.
# No hostname, DNS/UDP exemption or arbitrary loopback caller.
if (event == "socket.getaddrinfo" and not self.resolved
and frame.f_code is socket.getaddrinfo.__code__
and frame.f_back is not None and frame.f_back.f_code is code
and frame.f_back.f_locals.get("binds") == ["127.0.0.1"]
and frame.f_back.f_locals.get("port") == self.target_port
and frame.f_locals.get("host") == "127.0.0.1"
and frame.f_locals.get("port") == self.target_port
and frame.f_locals.get("family") == socket.AF_UNSPEC
and frame.f_locals.get("type") == socket.SOCK_STREAM
and frame.f_locals.get("proto") == 0
and frame.f_locals.get("flags") == socket.AI_PASSIVE
and args == ("127.0.0.1", self.target_port, socket.AF_UNSPEC, socket.SOCK_STREAM, 0)):
self.resolved = True
return
if (event == "socket.bind" and self.resolved and self.listener is None
and frame.f_code is code
and frame.f_locals.get("binds") == ["127.0.0.1"]
and frame.f_locals.get("port") == self.target_port
and args[1] == ("127.0.0.1", self.target_port)
and args[0] is frame.f_locals.get("sock")
and args[0].family == socket.AF_INET and args[0].type == socket.SOCK_STREAM):
self.listener = args[0]
Expand Down
8 changes: 6 additions & 2 deletions tests/test_cli_deployment.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,12 +98,16 @@ def create_app(*args, server_config, **kwargs):
return app
monkeypatch.setattr(server, "create_app", create_app)
launches = []
monkeypatch.setattr(web, "run_app", lambda app, **kwargs: launches.append((app, kwargs)))
def run(app, **kwargs):
assert "host" not in kwargs and "port" not in kwargs
assert [sock.getsockname()[1] for sock in kwargs["sock"]] == [9321]
launches.append((app, kwargs))
monkeypatch.setattr(web, "run_app", run)

cli.main()
assert configurations[0]["debug"] == {"profiling": True}
assert launches[0][0] is app
assert launches[0][1]["port"] == 9321
assert all(sock.fileno() == -1 for sock in launches[0][1]["sock"])
assert path.read_text() == original


Expand Down
5 changes: 3 additions & 2 deletions tests/test_cli_existing_owner.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,12 @@ def test_serve_refuses_existing_publishing_owner_without_traceback_or_takeover(t
home = tmp_path / 'app with spaces'
workspace = tmp_path / 'workspace'
workspace.mkdir()
# Reserve an unused address throughout the refusal. The second process must
# fail at publishing ownership before trying to bind any HTTP listener.
# Select an unused address, then let the CLI reserve its real listener before
# constructing the app. The separate publishing owner must still be refused.
with socket.socket() as reserved:
reserved.bind(('127.0.0.1', 0))
port = reserved.getsockname()[1]
reserved.close()
config = load_server_config(home)
save_server_config(home, {**config, 'port': port})
env = {**os.environ, 'AMPLIFIER_WEB_HOME': str(home),
Expand Down
Loading
Loading