diff --git a/src/solstone_tmux/cli.py b/src/solstone_tmux/cli.py index 8cb41a0..284e46b 100644 --- a/src/solstone_tmux/cli.py +++ b/src/solstone_tmux/cli.py @@ -45,13 +45,6 @@ def cmd_run(args: argparse.Namespace) -> int: config = load_config() config.ensure_dirs() - if not config.stream: - try: - config.stream = stream_name(host=socket.gethostname(), qualifier="tmux") - except ValueError as e: - print(f"Error: {e}", file=sys.stderr) - return 1 - if args.interval: config.segment_interval = args.interval @@ -81,54 +74,35 @@ def cmd_setup(args: argparse.Namespace) -> int: print("Error: journal URL is required", file=sys.stderr) return 1 - # Derive stream name - if not config.stream: - try: - config.stream = stream_name(host=socket.gethostname(), qualifier="tmux") - except ValueError as e: - print(f"Error deriving stream name: {e}", file=sys.stderr) - return 1 - print(f"Stream: {config.stream}") - # Save config before registration (so URL is persisted) config.ensure_dirs() save_config(config) - # Auto-register — try sol CLI first (no server needed), fall back to HTTP + # Register over HTTP — the journal mints the key and locks the stream. if not config.key: - sol = shutil.which("sol") - if sol: - print("Registering via sol CLI...") - try: - result = subprocess.run( - [sol, "observer", "--json", "create", config.stream], - capture_output=True, - text=True, - timeout=10, - ) - if result.returncode == 0: - data = json.loads(result.stdout) - config.key = data["key"] - save_config(config) - print(f"Registered (key: {config.key[:8]}...)") - else: - print("CLI registration failed, trying HTTP...") - except (subprocess.TimeoutExpired, json.JSONDecodeError, KeyError, OSError): - print("CLI registration failed, trying HTTP...") - - if not config.key: - print("Registering with your journal...") - client = UploadClient(config) - if client.ensure_registered(config): - config = load_config() - print(f"Registered (key: {config.key[:8]}...)") - else: - print( - "Warning: registration failed. Run setup again when your journal is reachable." - ) + print("Registering with your journal...") + client = UploadClient(config) + if client.ensure_registered(config): + config = load_config() + print(f"Registered (key: {config.key[:8]}...)") + else: + print( + "Warning: registration failed. Run setup again when your journal is reachable." + ) else: print(f"Already registered (key: {config.key[:8]}...)") + # The journal owns the stream once registered; derive a local fallback so + # capture still works before/without a reachable journal. + if not config.stream: + try: + config.stream = stream_name(host=socket.gethostname(), qualifier="tmux") + except ValueError as e: + print(f"Error deriving stream name: {e}", file=sys.stderr) + return 1 + save_config(config) + print(f"Stream: {config.stream}") + print(f"\nConfig saved to {config.config_path}") print(f"Captures will go to {config.captures_dir}") print( diff --git a/src/solstone_tmux/observer.py b/src/solstone_tmux/observer.py index f2bf586..3e67da2 100644 --- a/src/solstone_tmux/observer.py +++ b/src/solstone_tmux/observer.py @@ -63,6 +63,13 @@ class TmuxObserver: logger.error("Tmux not available") return False + self._client = UploadClient(self.config) + if self.config.server_url: + self._client.ensure_registered(self.config) + + # The journal owns the stream once registered; fall back to a local + # derivation so capture still works before/without a reachable journal. + self.stream = self.config.stream if not self.stream: try: self.stream = stream_name(host=HOST, qualifier="tmux") @@ -71,10 +78,6 @@ class TmuxObserver: logger.error(f"Failed to derive stream name: {e}") return False - self._client = UploadClient(self.config) - if self.config.server_url: - self._client.ensure_registered(self.config) - self._sync = SyncService(self.config, self._client) logger.info(f"Observer initialized: stream={self.stream}") return True @@ -182,7 +185,6 @@ class TmuxObserver: tmux=tmux_info, host=HOST, platform=PLATFORM, - stream=self.stream, ) if self.config.status_indicator and self._sync: indicator.update(self._sync.is_connected) diff --git a/src/solstone_tmux/sync.py b/src/solstone_tmux/sync.py index 39fa5ff..41d471a 100644 --- a/src/solstone_tmux/sync.py +++ b/src/solstone_tmux/sync.py @@ -18,7 +18,6 @@ import time import shutil from datetime import datetime, timedelta from pathlib import Path -from typing import Any from .config import Config from .upload import UploadClient @@ -327,14 +326,12 @@ class SyncService: if not files: return True # Nothing to upload - meta: dict[str, Any] = {"stream": self._config.stream} - retry_delays = self._config.sync_retry_delays max_retries = self._config.sync_max_retries for attempt in range(max_retries): result = await asyncio.to_thread( - self._client.upload_segment, day, segment_key, files, meta + self._client.upload_segment, day, segment_key, files ) if result.success: diff --git a/src/solstone_tmux/upload.py b/src/solstone_tmux/upload.py index 86a48c5..42f1a74 100644 --- a/src/solstone_tmux/upload.py +++ b/src/solstone_tmux/upload.py @@ -11,14 +11,15 @@ from __future__ import annotations import json import logging -import shutil -import subprocess +import platform +import socket import time from pathlib import Path from typing import Any, NamedTuple import requests +from . import __version__ from .config import Config logger = logging.getLogger(__name__) @@ -38,7 +39,6 @@ class UploadClient: def __init__(self, config: Config): self._url = config.server_url.rstrip("/") if config.server_url else "" self._key = config.key - self._stream = config.stream self._revoked = False self._session = requests.Session() self._retry_backoff = config.sync_retry_delays[:3] or [1, 5, 15] @@ -48,68 +48,52 @@ class UploadClient: def is_revoked(self) -> bool: return self._revoked - def _persist_key(self, config: Config, key: str) -> None: - """Save auto-registered key back to config.""" + def _persist_registration(self, config: Config, key: str, name: str) -> None: + """Save the minted key and journal-locked stream back to config.""" from .config import save_config config.key = key + config.stream = name save_config(config) def ensure_registered(self, config: Config) -> bool: - """Ensure the client has a valid key, auto-registering if needed. + """Ensure the client has a valid key, registering with the journal if needed. - Tries sol CLI first (no server needed), falls back to HTTP. + Registers HTTP-direct against /app/observer/register with a full + descriptor; the journal mints the key and locks the stream identity. Returns True if a key is available. """ if self._key: return True - # Try sol CLI registration first - name = self._stream or "solstone-tmux" - sol = shutil.which("sol") - if sol: - try: - result = subprocess.run( - [sol, "observer", "--json", "create", name], - capture_output=True, - text=True, - timeout=10, - ) - if result.returncode == 0: - data = json.loads(result.stdout) - self._key = data["key"] - self._persist_key(config, self._key) - logger.info(f"CLI-registered as '{name}' (key: {self._key[:8]}...)") - return True - except ( - subprocess.TimeoutExpired, - json.JSONDecodeError, - KeyError, - OSError, - ) as e: - logger.debug(f"CLI registration failed: {e}") - if not self._url: return False - url = f"{self._url}/app/observer/api/create" + descriptor = { + "platform": platform.system().lower(), + "hostname": socket.gethostname(), + "stream_type": "tmux", + "version": __version__, + } + url = f"{self._url}/app/observer/register" for attempt, delay in enumerate(self._retry_backoff): try: - resp = self._session.post( - url, json={"name": name}, timeout=EVENT_TIMEOUT - ) + resp = self._session.post(url, json=descriptor, timeout=EVENT_TIMEOUT) if resp.status_code == 200: data = resp.json() self._key = data["key"] - self._persist_key(config, self._key) + self._persist_registration(config, data["key"], data["name"]) logger.info( - f"Auto-registered as '{name}' (key: {self._key[:8]}...)" + f"Registered as '{data['name']}' (key: {self._key[:8]}...)" ) return True elif resp.status_code == 403: self._revoked = True - logger.error("Registration rejected (403)") + logger.error( + "Registration rejected — your journal must be reachable " + "directly on localhost" + ) return False else: logger.warning( @@ -134,7 +118,7 @@ class UploadClient: if self._revoked or not self._key or not self._url: return UploadResult(False) - url = f"{self._url}/app/observer/ingest/{self._key}" + url = f"{self._url}/app/observer/ingest" for attempt, delay in enumerate(self._retry_backoff): file_handles = [] @@ -158,7 +142,11 @@ class UploadClient: data["meta"] = json.dumps(meta) response = self._session.post( - url, data=data, files=files_data, timeout=UPLOAD_TIMEOUT + url, + data=data, + files=files_data, + headers={"Authorization": f"Bearer {self._key}"}, + timeout=UPLOAD_TIMEOUT, ) if response.status_code == 200: @@ -202,13 +190,14 @@ class UploadClient: if self._revoked or not self._key or not self._url: return None - url = f"{self._url}/app/observer/ingest/{self._key}/segments/{day}" - params = {} - if self._stream: - params["stream"] = self._stream + url = f"{self._url}/app/observer/ingest/segments/{day}" try: - resp = self._session.get(url, params=params, timeout=EVENT_TIMEOUT) + resp = self._session.get( + url, + headers={"Authorization": f"Bearer {self._key}"}, + timeout=EVENT_TIMEOUT, + ) if resp.status_code == 200: return resp.json() if resp.status_code in (401, 403): @@ -227,10 +216,15 @@ class UploadClient: if self._revoked or not self._key or not self._url: return False - url = f"{self._url}/app/observer/ingest/{self._key}/event" + url = f"{self._url}/app/observer/ingest/event" payload = {"tract": tract, "event": event, **fields} try: - resp = self._session.post(url, json=payload, timeout=EVENT_TIMEOUT) + resp = self._session.post( + url, + json=payload, + headers={"Authorization": f"Bearer {self._key}"}, + timeout=EVENT_TIMEOUT, + ) if resp.status_code == 200: return True if resp.status_code == 403: diff --git a/tests/test_upload.py b/tests/test_upload.py new file mode 100644 index 0000000..aadb3a7 --- /dev/null +++ b/tests/test_upload.py @@ -0,0 +1,136 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from unittest.mock import Mock + +import solstone_tmux +from solstone_tmux.config import Config, load_config +from solstone_tmux.upload import UploadClient + + +def test_register_posts_descriptor_and_persists(tmp_path): + config = Config(base_dir=tmp_path, server_url="http://localhost:5015") + config.ensure_dirs() + client = UploadClient(config) + client._session = Mock() + client._session.post = Mock( + return_value=Mock( + status_code=200, + json=lambda: { + "key": "abcd1234efgh", + "prefix": "abcd1234", + "name": "archon.tmux", + "ingest_url": "/app/observer/ingest", + "protocol_version": 2, + }, + ) + ) + + assert client.ensure_registered(config) is True + + call = client._session.post.call_args + assert call.args[0].endswith("/app/observer/register") + body = call.kwargs["json"] + assert body["stream_type"] == "tmux" + assert isinstance(body["platform"], str) and body["platform"] + assert body["hostname"] + assert body["version"] == solstone_tmux.__version__ + assert "stream" not in body and "name" not in body + assert config.key == "abcd1234efgh" and config.stream == "archon.tmux" + reloaded = load_config(tmp_path) + assert reloaded.key == "abcd1234efgh" and reloaded.stream == "archon.tmux" + + +def test_register_403_marks_revoked(tmp_path): + config = Config(base_dir=tmp_path, server_url="http://localhost:5015") + config.ensure_dirs() + client = UploadClient(config) + client._session = Mock() + client._session.post = Mock( + return_value=Mock(status_code=403, json=lambda: {}, text="") + ) + + assert client.ensure_registered(config) is False + assert client.is_revoked is True + assert client._session.post.call_count == 1 + + +def test_preset_key_skips_register(tmp_path): + config = Config( + base_dir=tmp_path, server_url="http://localhost:5015", key="EXISTINGKEY" + ) + config.ensure_dirs() + client = UploadClient(config) + client._session = Mock() + + assert client.ensure_registered(config) is True + client._session.post.assert_not_called() + + +def test_upload_segment_keyless_bearer_no_stream(tmp_path): + config = Config( + base_dir=tmp_path, server_url="http://localhost:5015", key="MYKEY123" + ) + config.ensure_dirs() + seg = tmp_path / "seg" + seg.mkdir() + f = seg / "tmux_main_screen.jsonl" + f.write_text("{}\n") + client = UploadClient(config) + client._session = Mock() + client._session.post = Mock( + return_value=Mock(status_code=200, json=lambda: {"status": "ok"}, text="") + ) + + res = client.upload_segment("20260610", "120000_300", [f]) + + assert res.success is True + call = client._session.post.call_args + assert call.args[0] == "http://localhost:5015/app/observer/ingest" + assert "/ingest/MYKEY123" not in call.args[0] + assert call.kwargs["headers"]["Authorization"] == "Bearer MYKEY123" + data = call.kwargs["data"] + assert "meta" not in data and "stream" not in data + + +def test_relay_event_keyless_bearer_no_stream(tmp_path): + config = Config( + base_dir=tmp_path, server_url="http://localhost:5015", key="MYKEY123" + ) + config.ensure_dirs() + client = UploadClient(config) + client._session = Mock() + client._session.post = Mock(return_value=Mock(status_code=200)) + + assert ( + client.relay_event("observe", "status", host="archon", platform="linux") is True + ) + + call = client._session.post.call_args + assert call.args[0].endswith("/app/observer/ingest/event") + assert "/ingest/MYKEY123" not in call.args[0] + assert call.kwargs["headers"]["Authorization"] == "Bearer MYKEY123" + payload = call.kwargs["json"] + assert payload["tract"] == "observe" and payload["event"] == "status" + assert "stream" not in payload + + +def test_get_server_segments_keyless_bearer_no_stream_param(tmp_path): + config = Config( + base_dir=tmp_path, server_url="http://localhost:5015", key="MYKEY123" + ) + config.ensure_dirs() + client = UploadClient(config) + client._session = Mock() + client._session.get = Mock( + return_value=Mock(status_code=200, json=lambda: [{"key": "120000_300"}]) + ) + + res = client.get_server_segments("20260610") + + assert res == [{"key": "120000_300"}] + call = client._session.get.call_args + assert call.args[0].endswith("/app/observer/ingest/segments/20260610") + assert "/ingest/MYKEY123" not in call.args[0] + assert call.kwargs["headers"]["Authorization"] == "Bearer MYKEY123" + assert "params" not in call.kwargs