Let the owner sign out all browsers at once

.github/workflows/platform-mechanisms.yml

@@ -89,6 +89,7 @@ on:
       - tests/test_update_checks.py
       - tests/test_windows_tls.py
       - tests/test_windows_trust.py
+      - tests/test_session_sign_out.py
       - tests/test_windows_service.py
       - tests/test_windows_task_runtime.py
       - tests/test_windows_installation.py
@@ -250,6 +251,11 @@ jobs:
         run: |
           .\.native-test-venv\Scripts\python.exe -B -m pytest -q -ra -s -p no:cacheprovider tests/test_windows_trust.py
           if ($LASTEXITCODE -ne 0) { throw 'Windows local-root trust checks failed.' }
+      - name: Check sign-out of all browsers on the real private store
+        shell: pwsh
+        run: |
+          .\.native-test-venv\Scripts\python.exe -B -m pytest -q -ra -p no:cacheprovider tests/test_session_sign_out.py
+          if ($LASTEXITCODE -ne 0) { throw 'Sign-out of all browsers checks failed.' }
   windows-user-trial:
     if: ${{ inputs.storage_only || (!inputs.bootstrap_only && !inputs.foreground_only) }}
     # Candidate-only combined boundary, not a universal suite/release gate.

amplifier_web/auth.py

@@ -161,22 +161,155 @@ def _signed(secret: str, payload: dict) -> str:
     return encoded + "." + signature
 
 
-def _verified(secret: str, value: str | None, ttl: int, kind: str) -> bool:
+def _issued(secret: str, value: str | None, ttl: int, kind: str) -> int | None:
+    """Return the signed issue second of a valid, unexpired value, else None."""
     if not value or "." not in value:
-        return False
+        return None
     encoded, signature = value.rsplit(".", 1)
     expected = hmac.new(secret.encode(), encoded.encode(), hashlib.sha256).hexdigest()
     if not hmac.compare_digest(signature, expected):
-        return False
+        return None
     try:
         payload = json.loads(base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)))
         issued = payload["issued"]
         if type(issued) is not int:
-            return False
+            return None
         age = time.time() - issued
-        return payload["kind"] == kind and 0 <= age <= ttl
+        return issued if payload["kind"] == kind and 0 <= age <= ttl else None
     except (OverflowError, ValueError, KeyError, TypeError, json.JSONDecodeError):
+        return None
+
+
+def _verified(secret: str, value: str | None, ttl: int, kind: str) -> bool:
+    return _issued(secret, value, ttl, kind) is not None
+
+
+def valid_session(app, value: str | None) -> bool:
+    """The one check for a browser session cookie.
+
+    A session from before the owner's last "sign out all browsers" is refused,
+    whatever its remaining lifetime. A session issued in the same second as
+    that sign-out is refused as well.
+    """
+    issued = _issued(app["session_secret"], value, app["server_config"]["session_ttl_seconds"], "session")
+    return issued is not None and issued > app.get("session_epoch", 0)
+
+
+_EPOCH = "session-epoch"
+
+
+def _epoch_path(data_dir: Path) -> Path:
+    if _WINDOWS:
+        from . import windows_files
+        return windows_files.address(data_dir) / "config" / "auth" / _EPOCH
+    return Path(data_dir).expanduser().resolve() / "config" / "auth" / _EPOCH
+
+
+def _read_epoch(data_dir: Path) -> int:
+    """Read the saved sign-out second. Missing means no sign-out yet (0)."""
+    try:
+        if _WINDOWS:
+            from . import windows_files
+            root = windows_files.address(data_dir)
+            with windows_files.directory(root / "config" / "auth", owned_root=root) as pinned:
+                content = pinned.read_private(_EPOCH, limit=32)
+        else:
+            content = _epoch_path(data_dir).read_bytes()[:33]
+    except FileNotFoundError:
+        return 0
+    if not re.fullmatch(rb"[0-9]{1,12}\n?", content):
+        raise ValueError("The saved sign-out time is malformed.")
+    return int(content)
+
+
+def _write_epoch(data_dir: Path, value: int) -> None:
+    if _WINDOWS:
+        write_private(_epoch_path(data_dir), f"{value}\n", owned_root=validated_data_dir(data_dir, create=True))
+    else:
+        auth_dir(data_dir)
+        write_private(_epoch_path(data_dir), f"{value}\n")
+
+
+def session_epoch(data_dir: Path) -> int:
+    """The second before which every browser session is refused.
+
+    An unreadable or malformed record never revives a session: it is replaced
+    by the present moment, once, so that everyone signs in again. A missing
+    record means no sign-out has happened. Anyone who could delete it can also
+    read the session secret, so its absence is not treated as an attack.
+    """
+    try:
+        return _read_epoch(data_dir)
+    except (OSError, ValueError):
+        now = int(time.time())
+        try:
+            _write_epoch(data_dir, now)
+        except OSError:
+            pass  # Still refused in memory; the next read repeats this.
+        return now
+
+
+def sign_out_all(data_dir: Path) -> int:
+    """Sign out every browser of this installation; return the saved second.
+
+    Authority is the owner's own private folder: no network request and no
+    credential are involved. A running host applies the change within a few
+    seconds. The time never moves backward, and a write failure is an error,
+    never a silent success.
+    """
+    try:
+        saved = _read_epoch(data_dir)
+    except (OSError, ValueError):
+        saved = 0
+    value = max(int(time.time()), saved)
+    _write_epoch(data_dir, value)
+    if _read_epoch(data_dir) != value:
+        raise OSError("The sign-out time was not saved.")
+    return value
+
+
+def _epoch_hint(data_dir: Path):
+    """A cheap change hint for the saved record; never its value."""
+    try:
+        info = os.stat(_epoch_path(data_dir))
+        return info.st_mtime_ns, info.st_size
+    except OSError:
+        return None
+
+
+def refresh_sessions(app, streams=()) -> bool:
+    """Apply a newer saved sign-out to the running host; True when applied.
+
+    Pending sign-in tickets are dropped, and open state streams are ended so
+    that a signed-out browser cannot keep watching. A stream that reconnects
+    must pass the session check again.
+    """
+    data_dir = app["service"].data_dir
+    hint = _epoch_hint(data_dir)
+    if hint == app.get("session_epoch_hint", ...):
         return False
+    app["session_epoch_hint"] = hint
+    value = session_epoch(data_dir)
+    if value <= app.get("session_epoch", 0):
+        return False
+    app["session_epoch"] = value
+    store = app.get("browser_entry")
+    if store is not None:
+        store.clear()
+    for task in list(streams):
+        task.cancel()
+    return True
+
+
+async def watch_sessions(app, streams, *, interval=2.0):
+    """Notice the owner's sign-out while the host runs."""
+    import asyncio
+    while True:
+        await asyncio.sleep(interval)
+        try:
+            refresh_sessions(app, streams)
+        except Exception:  # A watcher fault must never stop the host.
+            continue
 
 
 def new_session(secret: str) -> str:
@@ -329,7 +462,7 @@ async def authenticated(credential):
                     return web.json_response({'error': 'Device access was removed.'}, status=401)
                 return await handler(request)
         return web.json_response({'error': 'Device access was removed or is invalid.'}, status=401)
-    if _verified(request.app["session_secret"], request.cookies.get(SESSION_COOKIE), request.app["server_config"]["session_ttl_seconds"], "session"):
+    if valid_session(request.app, request.cookies.get(SESSION_COOKIE)):
         return await authenticated(request.cookies[SESSION_COOKIE])
     if is_api_request(request):
         return web.json_response({"error": "Authentication required."}, status=401)

amplifier_web/browser_entry.py

@@ -157,6 +157,11 @@ def issue(self, binding: Binding, next_path="/") -> str:
                 now + self.ttl, binding, validate_next_path(next_path))
             return ticket
 
+    def clear(self) -> None:
+        """Drop every pending ticket: a sign-out must not be followed by an old one."""
+        with self._lock:
+            self._grants.clear()
+
     def consume(self, ticket: str, binding: Binding) -> str | None:
         if not isinstance(ticket, str) or not re.fullmatch(r"[A-Za-z0-9_-]{43}", ticket):
             return None

amplifier_web/cli.py

@@ -97,6 +97,9 @@ def _parse() -> argparse.Namespace:
     runtime = subcommands.add_parser("runtime", help="Prepare or inspect the default installation runtime without a conversation")
     runtime.add_argument("runtime_command", choices=["prepare", "status"])
     runtime.add_argument("--json", action="store_true", help=argparse.SUPPRESS)
+    browsers = subcommands.add_parser("browsers", help="Manage signed-in browsers")
+    browsers.add_argument("browsers_command", choices=["sign-out"],
+                          help="sign-out: sign out every browser on every device")
     setup_tls = subcommands.add_parser("setup-tls", help="Create or inspect the app-owned local CA")
     setup_tls.add_argument("mode", nargs="?", choices=["status", "default", "force", "export"], default="default")
     reset = subcommands.add_parser("reset", help="Reinstall a clean app and runtime while preserving user data")
@@ -146,13 +149,13 @@ def _server_overrides(args: argparse.Namespace) -> dict:
 
 
 def _print_completion(shell: str) -> None:
-    words = "serve open tui acp run continue tool access doctor reset config service setup-tls completion --port --workspace --data-dir --no-open --bind --host --public-origin --tls-cert --tls-key --session-ttl --version"
+    words = "serve open tui acp run continue tool access doctor reset config service setup-tls browsers completion --port --workspace --data-dir --no-open --bind --host --public-origin --tls-cert --tls-key --session-ttl --version"
     if shell == "bash":
         print('complete -W "' + words + '" amplifier-unified')
     elif shell == "zsh":
-        print('#compdef amplifier-unified\n_arguments "1:command:(serve open tui acp run continue tool access doctor reset config service setup-tls)"')
+        print('#compdef amplifier-unified\n_arguments "1:command:(serve open tui acp run continue tool access doctor reset config service setup-tls browsers)"')
     else:
-        print('complete -c amplifier-unified -f -a "serve open tui acp run continue tool access doctor reset config service setup-tls"')
+        print('complete -c amplifier-unified -f -a "serve open tui acp run continue tool access doctor reset config service setup-tls browsers"')
 
 
 def _config(args, data_dir: Path) -> None:
@@ -503,6 +506,16 @@ def main():
     if args.command == "config":
         _config(args, data_dir)
         return
+    if args.command == "browsers":
+        from .auth import sign_out_all
+        try:
+            sign_out_all(data_dir)
+        except (OSError, ValueError):
+            raise SystemExit("Browsers were NOT signed out: the sign-out time could not be saved in "
+                             f"{data_dir}. Nothing was changed.") from None
+        print("All browsers on all devices are signed out. A running Amplifier applies this within a few seconds.")
+        print("To use Amplifier in a browser again, open it from this computer.")
+        return
     if args.command == "doctor":
         from .installation_health import doctor, display
         result = doctor(data_dir)

amplifier_web/device_connections.py

@@ -13,7 +13,7 @@
 
 from aiohttp import web
 
-from .auth import SESSION_COOKIE, _verified
+from .auth import SESSION_COOKIE, valid_session
 from .deployment import write_private
 from .device_capabilities import GRANTS, action, grants, transient
 from .terminal_devices import digest
@@ -420,7 +420,7 @@ def rotate(self, identity, rotation_id):
 
 def browser_cookie(request):
     value = request.cookies.get(SESSION_COOKIE)
-    if not _verified(request.app['session_secret'], value, request.app['server_config']['session_ttl_seconds'], 'session'):
+    if not valid_session(request.app, value):
         raise web.HTTPForbidden(text='Sign in with your browser to approve connections.')
     return value
 

amplifier_web/device_pairing.py

@@ -12,7 +12,7 @@
 
 from aiohttp import web
 
-from .auth import SESSION_COOKIE, _verified
+from .auth import SESSION_COOKIE, valid_session
 from .terminal_devices import MAX_DEVICES, digest
 from .terminal_setup import require_safe_transport
 
@@ -160,7 +160,7 @@ def setup_routes(app):
 
     def browser_session(request):
         cookie = request.cookies.get(SESSION_COOKIE)
-        if not _verified(app['session_secret'], cookie, app['server_config']['session_ttl_seconds'], 'session'):
+        if not valid_session(app, cookie):
             raise web.HTTPForbidden(text='Sign in with your browser to manage devices.')
         return cookie
 

amplifier_web/server.py

@@ -108,6 +108,9 @@ async def create_app(data_dir, workspace=None, runtime=None, voice=True, backgro
         for host in ("localhost", "127.0.0.1", "[::1]")
     )
     app["session_secret"] = session_secret(data_dir)
+    from .auth import _epoch_hint, session_epoch
+    app["session_epoch_hint"] = _epoch_hint(data_dir)
+    app["session_epoch"] = session_epoch(data_dir)
     app["control_token"] = control_token(data_dir)
     service = AppService(data_dir, runtime=runtime, workspace=workspace)
     service.port = config["port"]
@@ -838,6 +841,10 @@ async def index(request):
 
     async def shutdown(app):
         service.begin_shutdown()
+        watch = app.get("session_watch")
+        if watch is not None:
+            watch.cancel()
+            await asyncio.gather(watch, return_exceptions=True)
         waiting = list(streams | call_waiters | action_waiters)
         for task in waiting:
             task.cancel()
@@ -846,6 +853,8 @@ async def shutdown(app):
     app.on_shutdown.append(shutdown)
 
     async def confirm_started(app):
+        from .auth import watch_sessions
+        app["session_watch"] = asyncio.create_task(watch_sessions(app, streams))
         if background_work:
             # Resume only durable, positively unsubmitted shared-room work.
             # Configuration and runtime ownership are installed before attach.

docs/DEPLOYMENT.md

@@ -233,6 +233,30 @@ References: [Ubuntu trust-store guide](https://ubuntu.com/server/docs/how-to/sec
 and [Edge root-store policy history](https://learn.microsoft.com/en-us/deployedge/microsoft-edge-policies/microsoftrootstoreenabled).
 The obsolete Edge root-store policy is not a workaround.
 
+## Sign out all browsers
+
+To end every browser session on every device at once, for example after a
+phone is lost, run this on the host as the account that owns Unified:
+
+```sh
+amplifier-unified browsers sign-out
+```
+
+Pass the same `--data-dir` value if Unified uses a non-default data directory.
+The command saves a time in the owner's private configuration
+(`config/auth/session-epoch`). It needs no running host, no password and no
+network request: the ability to write that private folder is the authority.
+A running host applies the change within a few seconds. It then refuses every
+session from before that time, drops pending sign-in tickets, and ends open
+live-update connections so that a signed-out browser cannot keep watching.
+Sign in again in the usual way for your platform.
+
+Limits: this signs out **all** browsers; there is no sign-out for one device.
+Connected terminals, apps and devices with their own access are not browser
+sessions and stay connected; remove those in the app. The saved time never
+moves backward. If the record is damaged, the host signs everyone out once and
+repairs it.
+
 ## Keep the host running
 
 On Linux, including a WSL distribution with systemd enabled, install and manage

docs/design/native-windows-installation.md

@@ -1474,6 +1474,10 @@ This delegates existing OS-session authority; it does not prove a fresh Hello
 challenge. A new native Hello challenge is a different product/security promise.
 Keep PAM authentication unchanged for existing POSIX deployments.
 
+The owner can end every browser session with `amplifier-unified browsers sign-out`
+(see the deployment guide). All browser session checks go through one function,
+`auth.valid_session`, so that no check can ignore that sign-out.
+
 Remote browsers need an explicit owner-approved browser-pairing design over
 trusted configured HTTPS. Do not silently reuse terminal/companion grants with
 broader authority. Until that path is qualified, report its limitation rather

tests/test_server_auth.py

@@ -402,3 +402,38 @@ async def test_pwa_metadata_is_public_but_conversations_and_canvas_stay_private(
     login = await client.get('/login?error=1')
     assert login.headers['Cache-Control'] == 'no-store'
     assert 'Could not sign in.' in await login.text()
+
+
+async def test_real_host_applies_sign_out_all_browsers_to_requests_and_open_streams(aiohttp_client, tmp_path, monkeypatch):
+    import asyncio
+    from amplifier_web import auth
+    app = await create_app(tmp_path, preload_providers=False, workspace=tmp_path, runtime=Runtime(),
+                           voice=False, background_updates=False)
+    client = await aiohttp_client(app)
+    assert app["session_epoch"] == 0 and not app["session_watch"].done()
+    cookie = {"Cookie": f"amplifier_unified_session={new_session(app['session_secret'])}"}
+    assert (await client.get("/api/state", headers=cookie)).status == 200
+    stream = await client.get("/api/events", headers=cookie)
+    assert stream.status == 200
+    assert await stream.content.readany()  # The stream is open and delivering.
+    # The owner's command writes the private record; it sends the host nothing.
+    real = time.time
+    monkeypatch.setattr(auth.time, "time", lambda: real() + 5)
+    saved = auth.sign_out_all(tmp_path)
+    # The running host notices by itself (its watcher looks every two seconds).
+    for _ in range(100):
+        if app["session_epoch"] == saved:
+            break
+        await asyncio.sleep(0.1)
+    assert app["session_epoch"] == saved
+    # The same cookie is refused, for the API and for a page.
+    assert (await client.get("/api/state", headers=cookie)).status == 401
+    page = await client.get("/", headers=cookie, allow_redirects=False)
+    assert page.status == 307 and page.headers["Location"].startswith("/login")
+    # The open stream was ended by the host.
+    await asyncio.wait_for(stream.content.read(), 5)
+    assert stream.content.at_eof()
+    # A sign-in after the sign-out works.
+    monkeypatch.setattr(auth.time, "time", lambda: real() + 10)
+    fresh = {"Cookie": f"amplifier_unified_session={new_session(app['session_secret'])}"}
+    assert (await client.get("/api/state", headers=fresh)).status == 200