feat(auth): send X-GREFFON-TOKEN on cert poll and status callback

app/workers/monitor.py

@@ -28,6 +28,7 @@
 
 async def monitor_worker(app: FastAPI) -> None:
     settings: Settings = app.state.settings
+    token: str = app.state.greffer_token
     prev_status: dict[str, str] = {}
     try:
         while True:
@@ -40,6 +41,7 @@ async def monitor_worker(app: FastAPI) -> None:
                 await anyio.to_thread.run_sync(
                     _one_monitor_tick,
                     settings,
+                    token,
                     prev_status,
                     abandon_on_cancel=True,
                 )
@@ -55,7 +57,9 @@ async def monitor_worker(app: FastAPI) -> None:
         raise
 
 
-def _one_monitor_tick(settings: Settings, prev_status: dict[str, str]) -> None:
+def _one_monitor_tick(
+    settings: Settings, token: str, prev_status: dict[str, str]
+) -> None:
     """Run one monitoring pass. Mutates ``prev_status`` in place."""
     # Imported lazily so unit tests can mock before the docker SDK
     # initializes its from_env() client at import.
@@ -65,22 +69,22 @@ def _one_monitor_tick(settings: Settings, prev_status: dict[str, str]) -> None:
     for greffon_id in os.listdir(greffon_dir):
         status = compose.get_status(greffon_id)["status"]
         if prev_status.get(greffon_id) != status:
-            _report_status_change(settings, greffon_id, status)
+            _report_status_change(settings, token, greffon_id, status)
         prev_status[greffon_id] = status
 
 
 def _report_status_change(
-    settings: Settings, greffon_id: str, status: str
+    settings: Settings, token: str, greffon_id: str, status: str
 ) -> None:
-    """POST a status change to the manager.
-
-    Inlined from the now-deleted ``apps/utils/greffon/base_server.py``'s
-    ``change_status``. Six lines, exactly one caller — no reason for a
-    shared module anymore post-cutover.
+    """POST a status change to the manager. Carries ``X-GREFFON-TOKEN``
+    because the manager's greffon_status_changed endpoint authenticates
+    the greffer by that header and cross-checks ownership of the
+    instance id against the authenticated Greffer.
     """
     requests.post(
         f"{settings.greffon_base_server}/api/greffer/instances/{greffon_id}/",
         json={"status": status},
+        headers={"X-GREFFON-TOKEN": token},
         verify=settings.greffer_ssl_verify,
         timeout=_HTTP_TIMEOUT_SECONDS,
     )

app/workers/register.py

@@ -74,7 +74,7 @@ async def register_worker(app: FastAPI) -> None:
     while True:
         try:
             data = await anyio.to_thread.run_sync(
-                _fetch_cert, settings, abandon_on_cancel=True
+                _fetch_cert, settings, token, abandon_on_cancel=True
             )
         except (requests.ConnectionError, requests.Timeout):
             logger.info(
@@ -125,9 +125,15 @@ def _post_register(settings: Settings, address: str, token: str) -> None:
     )
 
 
-def _fetch_cert(settings: Settings) -> dict[str, Any] | None:
+def _fetch_cert(settings: Settings, token: str) -> dict[str, Any] | None:
+    # The manager's get_greffer_cert endpoint requires the X-GREFFON-TOKEN
+    # header (matched against Greffer.token or Greffer.new_token). The
+    # greffer sends the same token it posted to /register/ — accept_register
+    # copies new_token → token, so the value the greffer holds matches
+    # whichever field the manager looks up.
     res = requests.get(
         f"{settings.greffon_base_server}/api/greffer/certificate/{settings.greffer_id}/",
+        headers={"X-GREFFON-TOKEN": token},
         verify=settings.greffer_ssl_verify,
         timeout=_HTTP_TIMEOUT_SECONDS,
     )

tests/test_workers_monitor.py

@@ -41,9 +41,9 @@ def test_one_tick_calls_report_status_change_on_first_seen(
         "app.workers.monitor._report_status_change"
     ) as mock_report:
         mock_compose.get_status.return_value = {"status": "running"}
-        _one_monitor_tick(settings, prev)
+        _one_monitor_tick(settings, "tok", prev)
 
-    mock_report.assert_called_once_with(settings, "inst-a", "running")
+    mock_report.assert_called_once_with(settings, "tok", "inst-a", "running")
     assert prev == {"inst-a": "running"}
 
 
@@ -58,7 +58,7 @@ def test_one_tick_skips_report_status_change_when_unchanged(
         "app.workers.monitor._report_status_change"
     ) as mock_report:
         mock_compose.get_status.return_value = {"status": "running"}
-        _one_monitor_tick(settings, prev)
+        _one_monitor_tick(settings, "tok", prev)
 
     mock_report.assert_not_called()
 
@@ -74,9 +74,9 @@ def test_one_tick_fires_report_status_change_on_transition(
         "app.workers.monitor._report_status_change"
     ) as mock_report:
         mock_compose.get_status.return_value = {"status": "stopped"}
-        _one_monitor_tick(settings, prev)
+        _one_monitor_tick(settings, "tok", prev)
 
-    mock_report.assert_called_once_with(settings, "inst-a", "stopped")
+    mock_report.assert_called_once_with(settings, "tok", "inst-a", "stopped")
     assert prev == {"inst-a": "stopped"}
 
 
@@ -89,7 +89,7 @@ def test_report_status_change_posts_correct_payload(settings: Settings) -> None:
     from app.workers.monitor import _report_status_change
 
     with patch("app.workers.monitor.requests") as mock_requests:
-        _report_status_change(settings, "inst-42", "running")
+        _report_status_change(settings, "tok", "inst-42", "running")
 
     mock_requests.post.assert_called_once()
     url, = mock_requests.post.call_args.args
@@ -100,6 +100,19 @@ def test_report_status_change_posts_correct_payload(settings: Settings) -> None:
     assert kwargs["timeout"] == 10.0
 
 
+def test_report_status_change_sends_greffon_token_header(settings: Settings) -> None:
+    """Manager-side greffon_status_changed requires X-GREFFON-TOKEN. Without
+    this header the status callback 403s, and the dashboard silently shows
+    stale state."""
+    from app.workers.monitor import _report_status_change
+
+    with patch("app.workers.monitor.requests") as mock_requests:
+        _report_status_change(settings, "my-token", "inst-42", "running")
+
+    kwargs = mock_requests.post.call_args.kwargs
+    assert kwargs["headers"] == {"X-GREFFON-TOKEN": "my-token"}
+
+
 # ---------------------------------------------------------------------------
 # Async worker
 # ---------------------------------------------------------------------------
@@ -114,7 +127,7 @@ async def test_monitor_worker_continues_after_tick_exception(
 
     call_count = 0
 
-    def _fake_tick(_settings, _prev):
+    def _fake_tick(_settings, _token, _prev):
         nonlocal call_count
         call_count += 1
         if call_count == 1:
@@ -141,7 +154,7 @@ async def test_monitor_worker_cancellable_during_sleep(
     """Cancellation during asyncio.sleep propagates cleanly."""
     app = create_app(token="t", settings=settings)
 
-    def _noop_tick(_settings, _prev):
+    def _noop_tick(_settings, _token, _prev):
         return
 
     monkeypatch.setattr("app.workers.monitor._one_monitor_tick", _noop_tick)
@@ -168,7 +181,7 @@ async def test_monitor_worker_cancel_is_snappy_during_blocking_tick(
     tick_started = threading.Event()
     never_complete = threading.Event()
 
-    def _blocking_tick(_settings, _prev):
+    def _blocking_tick(_settings, _token, _prev):
         tick_started.set()
         never_complete.wait(timeout=10)
 

tests/test_workers_register.py

@@ -48,15 +48,29 @@ def test_fetch_cert_returns_data_on_200(settings: Settings) -> None:
         mock_response.status_code = 200
         mock_response.json.return_value = {"certificate": "c", "private_key": "k"}
         mock_requests.get.return_value = mock_response
-        assert _fetch_cert(settings) == {"certificate": "c", "private_key": "k"}
+        assert _fetch_cert(settings, "tok") == {"certificate": "c", "private_key": "k"}
 
 
 def test_fetch_cert_returns_none_on_non_200(settings: Settings) -> None:
     with patch("app.workers.register.requests") as mock_requests:
         mock_response = MagicMock()
         mock_response.status_code = 401
         mock_requests.get.return_value = mock_response
-        assert _fetch_cert(settings) is None
+        assert _fetch_cert(settings, "tok") is None
+
+
+def test_fetch_cert_sends_greffon_token_header(settings: Settings) -> None:
+    """The manager's get_greffer_cert endpoint requires X-GREFFON-TOKEN.
+    Without this header the post-accept cert poll would 403 against a
+    mTLS-hardened manager."""
+    with patch("app.workers.register.requests") as mock_requests:
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"certificate": "c", "private_key": "k"}
+        mock_requests.get.return_value = mock_response
+        _fetch_cert(settings, "my-token")
+    kwargs = mock_requests.get.call_args.kwargs
+    assert kwargs["headers"] == {"X-GREFFON-TOKEN": "my-token"}
 
 
 def test_install_cert_copies_files(settings: Settings) -> None:
@@ -255,7 +269,7 @@ def test_fetch_cert_carries_timeout(settings: Settings) -> None:
         mock_response.status_code = 200
         mock_response.json.return_value = {}
         mock_requests.get.return_value = mock_response
-        _fetch_cert(settings)
+        _fetch_cert(settings, "tok")
     assert "timeout" in mock_requests.get.call_args.kwargs
     assert mock_requests.get.call_args.kwargs["timeout"] == 10.0