fix(orchestrator): accept job events only from the job's own workers

CLAUDE.md

@@ -110,7 +110,8 @@ phased delivery plan; new work should move the code toward it, not entrench the
   fields use `core.fields.EncryptedJSONField`.
 - Backups are `backups.models.BackupJob`s: create them with `backups.jobs.create_job`, and let
   worker events drive their state (`backups.jobs.handle_job_event`). Never trust a worker's
-  claim without checking its signature and approval.
+  claim without checking its signature, its approval and that the job is its own (a collector of the device's
+  collection group, or a storage worker of the job's target).
 - Retention deletes stored content only through `backups.retention` (signed `storage.delete`, confirmed by a
   `deleted` event); a pruned job row stays as the record. Anything that deletes backups must skip pinned ones
   (`BackupJob.pinned`) and keep the newest `min_backups`. Two exceptions (`devices.lifecycle`): past an inactive device's hold, retention

docs/ARCHITECTURE.md

@@ -51,7 +51,7 @@ No component talks to another directly. Everything goes over AMQP or Redis. The
 | `events` | active/active (competing consumers) | Celery worker on `dv.orchestrator`. Handles `worker.register`, `job.collected`, `job.stored`, `job.failed`. Replaces `consume_device_results`, `consume_storage_results`, and `monitor_worker_registry`. |
 | `scheduler` | runs everywhere, **one leader** (Redis lock with TTL and renewal) | Reads `BackupSchedule` from the DB and dispatches due jobs. Sends liveness pings, times out stale jobs, and applies retention policies. Replaces celery beat, whose local schedule file isn't HA-safe. |
 
-Both are in place since phase 5: `events` handles worker registration, heartbeats and the signed job events (`orchestrator.job_event`: started, retrying, collected, stored, failed); `scheduler` holds the Redis lease `dv:scheduler:leader` (60 s or twice the tick interval, renewed every tick, released on shutdown) and creates `BackupJob`s for due schedules. Backup content travels collector → Redis (`dv:blob:job:<id>`, 24 h TTL) → storage worker; a job is `stored` only when the collector and storage worker report the same SHA-256.
+Both are in place since phase 5: `events` handles worker registration, heartbeats and the signed job events (`orchestrator.job_event`: started, retrying, collected, stored, failed); `scheduler` holds the Redis lease `dv:scheduler:leader` (60 s or twice the tick interval, renewed every tick, released on shutdown) and creates `BackupJob`s for due schedules. Backup content travels collector → Redis (`dv:blob:job:<id>`, 24 h TTL) → storage worker; a job is `stored` only when the collector and storage worker report the same SHA-256. Job events are accepted only from the job's own workers: the storing step from a storage worker of the job's target, everything else from a collector of the device's collection group (a pushed backup has no collecting step to report). An approved worker's signature on an event about any other job changes nothing and is logged (`job_event_refused`).
 
 ### Worker runtime (pure Celery + kombu, **no Django**)
 

orchestrator/backups/jobs.py

@@ -186,6 +186,14 @@ def handle_job_event(envelope: Any) -> BackupJob | None:
             _apply_deletion(job, payload, worker)
             job.save()
             return job
+        refusal = _not_its_worker(job, payload, worker)
+        if refusal:
+            # Signed by an approved worker, but about a job that is none of its business.
+            logger.warning(
+                'job_event_refused',
+                extra={'job_id': str(job.id), 'worker': worker.name, 'event': payload.get('event'), 'reason': refusal},
+            )
+            return job
         was = job.state
         _apply(job, payload, worker)
         job.save()
@@ -196,6 +204,26 @@ def handle_job_event(envelope: Any) -> BackupJob | None:
     return job
 
 
+def _not_its_worker(job: BackupJob, payload: dict[str, Any], worker: Worker) -> str | None:
+    """Why ``worker`` may not report on ``job`` (None if it may).
+
+    A signature says who sent an event, not that the job is theirs. The storing step belongs to the
+    storage workers of the job's target; everything else to the collectors of the device's collection
+    group. A pushed backup has no collecting step to report on: the orchestrator created it collected.
+    Membership is read now, so a worker taken off a group or target mid-job loses its say in it.
+    """
+    if payload.get('event') == EVENT_STORED or payload.get('stage') == STAGE_STORE:
+        if worker.role == ROLE_STORAGE and worker.storage_targets.filter(pk=job.storage_target_id).exists():
+            return None
+        return "not a storage worker of the job's target"
+    if job.trigger == BackupJob.TRIGGER_PUSH:
+        return 'a pushed backup is not collected by a worker'
+    group_id = job.device.collection_group_id
+    if worker.role == ROLE_COLLECTOR and group_id and worker.collection_groups.filter(pk=group_id).exists():
+        return None
+    return "not a collector of the device's collection group"
+
+
 def _apply(job: BackupJob, payload: dict[str, Any], worker: Worker) -> None:
     event = payload.get('event')
     now = timezone.now()
@@ -249,14 +277,14 @@ def _apply(job: BackupJob, payload: dict[str, Any], worker: Worker) -> None:
 def _pin_host_key(job: BackupJob, payload: dict[str, Any], worker: Worker) -> None:
     """Pin the SSH host key a device presented at its first collected backup (trust on first use).
 
-    Only one of the device's own collectors can pin it: an approved worker with no business
-    with the device must not be able to choose the key its collectors will then trust.
+    The event comes from one of the device's own collectors (``_not_its_worker``), so no other
+    approved worker can choose the key they will then trust.
     """
     from audit.log import record
     from devices.models import Device
 
     key, device = payload.get('host_key'), job.device
-    if not key or job.trigger == BackupJob.TRIGGER_PUSH:
+    if not key:
         return
     fingerprint = host_key_fingerprint(key)
     if fingerprint is None:
@@ -267,10 +295,6 @@ def _pin_host_key(job: BackupJob, payload: dict[str, Any], worker: Worker) -> No
             # Its collectors enforce the pin, so this is an old job or a collector that does not.
             logger.warning('host_key_ignored', extra={'job_id': str(job.id), 'reason': 'another key is pinned'})
         return
-    serves = worker.role == ROLE_COLLECTOR and worker.collection_groups.filter(pk=device.collection_group_id).exists()
-    if not serves:
-        logger.warning('host_key_ignored', extra={'job_id': str(job.id), 'reason': "not the device's collector"})
-        return
     pinned = Device.objects.filter(pk=device.pk, ssh_host_key='').update(
         ssh_host_key=key, ssh_host_key_pinned_at=timezone.now()
     )

orchestrator/backups/tests/test_jobs.py

@@ -305,3 +305,105 @@ def test_malformed_host_keys_are_not_pinned(setup, broker, key):
 
     setup['device'].refresh_from_db()
     assert setup['device'].ssh_host_key == ''
+
+
+# --- events are accepted only from the job's own workers -----------------------------------------
+
+
+@pytest.fixture
+def outsiders(setup, worker_keys):
+    """Approved workers with no part in the device's backups: a collector of another group, a storage
+    worker of another target, and a receiver."""
+    from core import workers
+    from devices.models import CollectionGroup
+    from storage.models import StorageTarget
+
+    collector, storage, receiver = (worker_keys(name) for name in ('collector-lab', 'storage-lab', 'receiver-1'))
+    lab = CollectionGroup.objects.create(name='Lab', rabbitmq_queue_id='lab')
+    workers.approve(workers.handle_event(collector.event()), role='collector', collection_groups=[lab])
+    target = StorageTarget.objects.create(name='Lab store', driver='filesystem', config={'path': 'lab'})
+    workers.approve(workers.handle_event(storage.event()), role='storage', storage_targets=[target])
+    workers.approve(workers.handle_event(receiver.event()), role='receiver')
+    return {'collector': collector, 'storage': storage, 'receiver': receiver}
+
+
+def unchanged(job):
+    fresh = BackupJob.objects.get(pk=job.pk)
+    return (fresh.state, fresh.log, fresh.error, fresh.collected_sha256, fresh.stored_sha256, fresh.storage_ref)
+
+
+@pytest.mark.parametrize('who', ['collector', 'storage', 'receiver'])
+@pytest.mark.parametrize(
+    'event',
+    [
+        {'event': EVENT_STARTED, 'attempt': 1},
+        {'event': EVENT_COLLECTED, 'sha256': SHA, 'size': 10},
+        {'event': EVENT_STORED, 'sha256': SHA, 'storage_ref': 'elsewhere'},
+        {'event': EVENT_STARTED, 'stage': STAGE_STORE},
+        {'event': EVENT_FAILED, 'error': 'forged'},
+        {'event': EVENT_FAILED, 'stage': STAGE_STORE, 'error': 'forged'},
+        {'event': 'made-up'},
+    ],
+)
+def test_workers_with_no_part_in_a_job_cannot_report_on_it(setup, broker, outsiders, who, event):
+    job, _ = create_job(setup['device'])
+    before = unchanged(job)
+    details = {key: value for key, value in event.items() if key != 'event'}
+
+    handle_job_event(job_event(outsiders[who], job, event['event'], log=[{'message': 'forged'}], **details))
+
+    assert unchanged(job) == before
+    setup['device'].refresh_from_db()
+    assert setup['device'].last_backup_status == ''
+
+
+def test_each_of_a_jobs_workers_reports_only_its_own_step(setup, broker):
+    job, _ = create_job(setup['device'])
+    before = unchanged(job)
+
+    # The collector cannot report the storing step, nor the storage worker the collecting one.
+    handle_job_event(job_event(setup['collector'], job, EVENT_STORED, sha256=SHA, storage_ref='ref'))
+    handle_job_event(job_event(setup['collector'], job, EVENT_FAILED, stage=STAGE_STORE, error='forged'))
+    handle_job_event(job_event(setup['storage'], job, EVENT_COLLECTED, sha256=SHA, size=10))
+    handle_job_event(job_event(setup['storage'], job, EVENT_FAILED, error='forged'))
+
+    assert unchanged(job) == before
+
+
+def test_a_finished_jobs_log_is_closed_to_other_workers(setup, broker, outsiders):
+    job, _ = create_job(setup['device'])
+    collected(setup, job)
+    job = stored(setup, job)
+    before = unchanged(job)
+
+    handle_job_event(job_event(outsiders['collector'], job, EVENT_STARTED, log=[{'message': 'forged'}]))
+
+    assert unchanged(job) == before
+
+
+def test_a_worker_taken_off_the_job_loses_its_say(setup, broker):
+    job, _ = create_job(setup['device'])
+    setup['group'].workers.clear()
+    setup['target'].workers.clear()
+
+    collected(setup, job)
+    stored(setup, job)
+
+    assert BackupJob.objects.get(pk=job.pk).state == BackupJob.QUEUED
+
+
+def test_no_worker_reports_the_collection_of_a_pushed_backup(setup, broker):
+    job = BackupJob.objects.create(
+        device=setup['device'],
+        state=BackupJob.COLLECTED,
+        trigger=BackupJob.TRIGGER_PUSH,
+        storage_target=setup['target'],
+        collected_sha256=SHA,
+    )
+
+    handle_job_event(job_event(setup['collector'], job, EVENT_COLLECTED, sha256='b' * 64, size=10))
+    handle_job_event(job_event(setup['collector'], job, EVENT_FAILED, error='forged'))
+    assert BackupJob.objects.get(pk=job.pk).collected_sha256 == SHA
+
+    # Its storage step is still the storage worker's to report.
+    assert stored(setup, job).state == BackupJob.STORED

protocol/src/devicevault_protocol/jobs.py

@@ -31,12 +31,17 @@
 The orchestrator only accepts a ``stored`` event whose hash matches the one
 the collector reported.
 
+An event is accepted only from a worker the job belongs to: the storing step
+(``stored``, and any event with ``stage`` "store") from a storage worker of
+the job's target, everything else from a collector of the device's collection
+group. Events about a pushed backup's collection are refused, since no worker
+collects it.
+
 SSH host keys are pinned on first use. ``job.collect`` carries the device's
 pinned key as ``device.host_key`` (``"<type> <base64>"``, as in known_hosts, or
 None); a collector refuses a device presenting any other key (reason ``auth``).
 Every ``collected`` event reports the key the device presented, and the
-orchestrator pins it if the device has none yet and the event comes from one
-of the device's own collectors.
+orchestrator pins it if the device has none yet.
 
 Retention: the orchestrator sends ``storage.delete`` (signed) to a storage
 target's queue with ``items`` (``[{"job_id", "storage_ref"}, ...]``); the