Use the normal host PAM account for isolated Unified sign-in

amplifier_web/auth.py

@@ -356,7 +356,14 @@ async def post_login(request: web.Request) -> web.Response:
     if browser_entry.is_windows():
         return web.Response(text=browser_entry.REOPEN, status=403)
     form = await request.post()
-    if not authenticate_pam(str(form.get("username", "")), str(form.get("password", ""))):
+    username, password = str(form.get("username", "")), str(form.get("password", ""))
+    bridge = request.app["server_config"].get("host_pam")
+    if bridge is not None:
+        from .host_pam import authenticate
+        accepted = await authenticate(bridge, username, password)
+    else:
+        accepted = authenticate_pam(username, password)
+    if not accepted:
         url = login_url(request.query.get("next"))
         raise web.HTTPSeeOther(url + ("&" if "?" in url else "?") + "error=1")
     response = web.HTTPSeeOther(validate_next_path(request.query.get("next")))

amplifier_web/deployment.py

@@ -21,6 +21,7 @@
     "port": 8941,
     "public_origins": [],
     "session_ttl_seconds": 604800,
+    "host_pam": None,
     "tls": {"method": "none", "cert": "", "key": ""},
     "runtime": DEFAULT_RETENTION,
     "debug": {"profiling": False},
@@ -143,6 +144,8 @@ def validate_server(value: dict) -> dict:
         raise ValueError("port must be an integer from 1 through 65535.")
     if type(config["session_ttl_seconds"]) is not int or config["session_ttl_seconds"] <= 0:
         raise ValueError("session_ttl_seconds must be a positive integer.")
+    from .host_pam import validate as validate_host_pam
+    config["host_pam"] = validate_host_pam(config["host_pam"])
     origins = config["public_origins"]
     if not isinstance(origins, list):
         raise ValueError("public_origins must be a list.")

amplifier_web/host_pam.py

@@ -0,0 +1,54 @@
+"""Optional single-owner login through Constellation's private host PAM socket."""
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+from pathlib import Path
+import stat
+
+TIMEOUT_SECONDS = 20
+
+
+def validate(value):
+    if value is None:
+        return None
+    if (os.name == "nt" or not isinstance(value, dict)
+            or set(value) != {"owner", "socket"}):
+        raise ValueError("host_pam requires owner and socket on a Unix host, or null.")
+    owner, path = value["owner"], value["socket"]
+    if (not isinstance(owner, str) or not owner or owner != owner.strip()
+            or len(owner) > 256 or any(ord(char) < 32 for char in owner)):
+        raise ValueError("host_pam.owner must name one host account.")
+    if (not isinstance(path, str) or not path or "\0" in path
+            or not Path(path).is_absolute()):
+        raise ValueError("host_pam.socket must be an absolute Unix socket path.")
+    return {"owner": owner, "socket": path}
+
+
+async def authenticate(config: dict, username: str, password: str) -> bool:
+    """Use the existing owner-login.sock protocol; never retain credentials.
+
+    The installing owner supplies this private local socket. The host bridge
+    applies its fixed-owner PAM policy, account checks and attempt limits.
+    Unavailable or invalid bridges fail closed, without a local PAM fallback.
+    """
+    if username != config["owner"] or not password or len(password) > 1024:
+        return False
+    writer = None
+    try:
+        info = Path(config["socket"]).lstat()
+        if (not stat.S_ISSOCK(info.st_mode) or info.st_mode & 0o077
+                or info.st_uid not in {0, os.geteuid()}):
+            return False
+        async with asyncio.timeout(TIMEOUT_SECONDS):
+            reader, writer = await asyncio.open_unix_connection(config["socket"], limit=64)
+            writer.write((json.dumps({"username": username, "password": password},
+                                     ensure_ascii=False) + "\n").encode("utf-8"))
+            await writer.drain()
+            return await reader.readline() == b"OK\n" and await reader.read(1) == b""
+    except (OSError, ValueError, TimeoutError):
+        return False
+    finally:
+        if writer is not None:
+            writer.close()

docs/DEPLOYMENT.md

@@ -72,6 +72,42 @@ hostname.
 
 If Unified uses a non-default data directory, pass the same `--data-dir` value to `setup-tls` and `doctor`.
 
+## Sign in with the installing host owner's account
+
+On Unix, the default login checks PAM for the account running Unified. An
+isolated service can instead use the installing human's existing host account
+through the [Constellation owner-login bridge](https://github.com/microsoft/amplifier-constellation-instance/blob/main/ops/container/README.md).
+This keeps the service's OS identity, data directory and conversation ownership
+separate from the account checking the password.
+
+Run the canonical bridge as the intended host owner, with a private installation
+configuration and socket directory. Expose **only its `owner-login.sock`** through
+the container runtime's private Unix-socket forwarding. The guest socket must be
+owned by root or the Unified service account, with no group/other permissions,
+inside a private directory. Do not mount the host password database, copy a
+password, expose this socket on TCP, or share another installation's bridge.
+No optional bridge TCP routes are needed for login.
+
+Configure the existing Unified installation, then restart it at a safe boundary:
+
+```sh
+amplifier-unified --data-dir /path/to/instance config set host_pam '{owner: your-host-account, socket: /run/your-installation/owner-login.sock}'
+```
+
+The normal HTTPS login page accepts that one host username and its normal PAM
+password. The canonical bridge applies host authentication, account policy and
+attempt limits. A missing, unavailable or rejecting bridge denies the login;
+there is no fallback to the guest password. The app never saves or logs the
+submitted password. TLS, Origin, CSRF, signed-session cookies and control-token
+rules remain unchanged. This grants access to the same single-owner installation;
+it does not create a multi-user workspace, impersonate the user for tools, or
+change filesystem/agent permissions.
+
+Set `host_pam` to `null` and restart to restore the default process-owner PAM
+login. Windows keeps its existing native sign-in and does not accept this option.
+Existing browser sessions retain their normal lifetime through this setting
+change; changing the login backend is not a session-revocation operation.
+
 ## Trust the CA on each client
 
 On the Unified host, export the public CA and record the fingerprint:

tests/test_host_pam.py

@@ -0,0 +1,139 @@
+"""Single-owner host login, with a disposable peer for the existing bridge protocol."""
+import asyncio
+import json
+import os
+from pathlib import Path
+import tempfile
+from types import SimpleNamespace
+
+from aiohttp import web
+import pytest
+
+from amplifier_web import auth, host_pam
+from amplifier_web.deployment import validate_server, setting_set
+
+pytestmark = pytest.mark.skipif(os.name == "nt", reason="Unix socket login is opt-in on Unix only")
+
+
+@pytest.fixture
+async def bridge():
+    # Linux Unix socket paths are limited to 108 bytes; pytest's case directory
+    # names can exceed that before the socket filename is appended.
+    with tempfile.TemporaryDirectory(prefix="unified-pam-") as root:
+        path = Path(root) / "owner.sock"
+        state = SimpleNamespace(requests=[], reply=b"OK\n", delay=0, tasks=set())
+
+        async def connection(reader, writer):
+            state.tasks.add(asyncio.current_task())
+            try:
+                state.requests.append(json.loads(await reader.readline()))
+                await asyncio.sleep(state.delay)
+                writer.write(state.reply)
+                await writer.drain()
+            finally:
+                writer.close()
+                await writer.wait_closed()
+                state.tasks.discard(asyncio.current_task())
+
+        server = await asyncio.start_unix_server(connection, path)
+        path.chmod(0o600)
+        try:
+            yield {"owner": "host-owner", "socket": str(path)}, state
+        finally:
+            server.close()
+            await server.wait_closed()
+            for task in tuple(state.tasks):
+                task.cancel()
+            await asyncio.gather(*state.tasks, return_exceptions=True)
+
+
+async def test_one_host_owner_and_exact_password_reach_the_bridge(bridge):
+    config, state = bridge
+    password = "synthetic café\n\"password\""
+    assert await host_pam.authenticate(config, "host-owner", password)
+    assert state.requests == [{"username": "host-owner", "password": password}]
+    assert not await host_pam.authenticate(config, "guest-service", password)
+    assert not await host_pam.authenticate(config, "host-owner", "")
+    assert not await host_pam.authenticate(config, "host-owner", "x" * 1025)
+    assert len(state.requests) == 1
+
+
+@pytest.mark.parametrize("reply", [b"NO\n", b"", b"OK", b"OK\nextra", b"x" * 80 + b"\n"])
+async def test_rejection_and_malformed_responses_fail_closed(bridge, reply):
+    config, state = bridge
+    state.reply = reply
+    assert not await host_pam.authenticate(config, "host-owner", "synthetic")
+
+
+async def test_slow_bridge_is_bounded_and_does_not_block_the_event_loop(bridge, monkeypatch):
+    config, state = bridge
+    state.delay = 1
+    monkeypatch.setattr(host_pam, "TIMEOUT_SECONDS", 0.02)
+    attempt = asyncio.create_task(host_pam.authenticate(config, "host-owner", "synthetic"))
+    await asyncio.sleep(0.005)
+    assert not attempt.done()
+    assert not await attempt
+
+
+async def test_missing_unshared_and_symlink_sockets_fail_closed(bridge, tmp_path):
+    config, state = bridge
+    assert not await host_pam.authenticate({**config, "socket": str(tmp_path / "absent")}, "host-owner", "synthetic")
+    path = Path(config["socket"])
+    path.chmod(0o666)
+    assert not await host_pam.authenticate(config, "host-owner", "synthetic")
+    path.chmod(0o600)
+    alias = path.with_name("alias")
+    alias.symlink_to(path)
+    assert not await host_pam.authenticate({**config, "socket": str(alias)}, "host-owner", "synthetic")
+    assert state.requests == []
+
+
+def test_opt_in_configuration_is_atomic_and_defaults_remain_local():
+    config = validate_server({})
+    assert config["host_pam"] is None
+    selected = setting_set(config, "host_pam", {"owner": "host-owner", "socket": "/run/private/pam.sock"})
+    assert selected["host_pam"]["owner"] == "host-owner"
+    assert config["host_pam"] is None
+    assert setting_set(selected, "host_pam", None)["host_pam"] is None
+    for invalid in ({}, {"owner": "", "socket": "/socket"},
+                    {"owner": "user", "socket": "relative"},
+                    {"owner": "user", "socket": "/socket", "url": "https://peer"},
+                    {"owner": "user\n", "socket": "/socket"}):
+        with pytest.raises(ValueError, match="host_pam"):
+            setting_set(config, "host_pam", invalid)
+
+
+async def test_normal_login_keeps_csrf_cookie_and_private_route_boundary(aiohttp_client, bridge, monkeypatch):
+    config, state = bridge
+    app = web.Application(middlewares=[auth.auth_required])
+    app["server_config"] = validate_server({"host_pam": config})
+    app["session_secret"] = "synthetic-session-key"
+    app["allowed_origins"] = set()
+    app["control_token"] = "synthetic-control"
+    app["service"] = SimpleNamespace(data_dir=Path("/synthetic"))
+    app.router.add_post("/login", auth.post_login)
+    async def private(_):
+        return web.Response(text="signed in")
+    app.router.add_get("/", private)
+    def no_guest_fallback(*_):
+        pytest.fail("Selected host login must not fall back to guest PAM")
+    monkeypatch.setattr(auth, "authenticate_pam", no_guest_fallback)
+    client = await aiohttp_client(app)
+    form = {"username": "host-owner", "password": "synthetic"}
+    assert (await client.post("/login", data=form)).status == 403
+    csrf = auth.new_csrf(app["session_secret"])
+    headers = {"Cookie": f"{auth.CSRF_COOKIE}={csrf}"}
+    form["csrf"] = csrf
+    rejected = await client.post("/login", data=form, headers={**headers, "Origin": "https://wrong.example"})
+    assert rejected.status == 403 and state.requests == []
+    state.reply = b"NO\n"
+    rejected = await client.post("/login", data=form, headers=headers, allow_redirects=False)
+    assert rejected.status == 303 and "error=1" in rejected.headers["Location"]
+    assert auth.SESSION_COOKIE not in rejected.cookies
+    state.reply = b"OK\n"
+    accepted = await client.post("/login", data=form, headers=headers, allow_redirects=False)
+    assert accepted.status == 303 and accepted.headers["Location"] == "/"
+    cookie = accepted.cookies[auth.SESSION_COOKIE]
+    assert cookie["httponly"] and cookie["samesite"] == "Strict"
+    assert auth._verified(app["session_secret"], cookie.value, 60, "session")
+    assert (await client.get("/", headers={"Cookie": f"{auth.SESSION_COOKIE}={cookie.value}"})).status == 200