diff --git a/CLAUDE.md b/CLAUDE.md
index 097ba69..8d3a185 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -99,7 +99,8 @@ phased delivery plan; new work should move the code toward it, not entrench the
 - Collector drivers subclass `CollectorDriver` (`worker/src/devicevault_worker/drivers/base.py`)
   with a pydantic `Params` model, and raise `TransientError` (retried) or `PermanentError`.
   SSH CLI platforms subclass `drivers.ssh.NetmikoDriver` (a Netmiko `device_type`, the
-  commands, and regexes for volatile lines); see `drivers/cisco.py`. Built-ins are listed in
+  commands, and regexes for volatile lines); see `drivers/cisco.py`. A driver that connects over SSH must
+  require `target.host_key` when it is set and report the key it saw in `log.host_key` (pinned on first use). Built-ins are listed in
   `drivers/__init__.py`; third-party drivers register under the
   `devicevault.collectors` entry point. Storage drivers subclass `storage.drivers.base.StorageDriver` (see `storage/drivers/local.py`, `s3.py`), stream
   content (`store` reads a binary file, `read` writes one; backups can be hundreds of MB) and register under
diff --git a/deploy/orchestrator/orchestrator.env.example b/deploy/orchestrator/orchestrator.env.example
index 79c619d..2cae1f4 100644
--- a/deploy/orchestrator/orchestrator.env.example
+++ b/deploy/orchestrator/orchestrator.env.example
@@ -9,6 +9,10 @@ DV_DEBUG=false
 DV_ALLOWED_HOSTS=devicevault.example.com
 # Full origins, required when the UI is served over HTTPS by a front proxy
 DV_CSRF_TRUSTED_ORIGINS=https://devicevault.example.com
+# Address (or network, comma separated) of the TLS proxy or load balancer in front of the
+# web service. Without it every user appears to come from the proxy: one client's failed
+# sign-ins lock everyone out, and API tokens' allowed networks check the proxy's address.
+DV_TRUSTED_PROXIES=
 DV_TIMEZONE=Australia/Sydney
 DV_LOG_LEVEL=INFO
 
diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index 98d95b0..0e9d790 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -101,6 +101,8 @@ Message formats are in `protocol/src/devicevault_protocol/registration.py`.
 
 **Trust on first use:** the worker pins the orchestrator key from the first *approved* assignment that verifies against it, so the trust decision coincides with the admin's approval. After that, assignments signed with any other key are rejected, and so is every task (`device.collect`, `storage.store`, `storage.read`) not signed by the pinned key. Tasks rejected this way fail immediately and are not retried. The Workers page shows the orchestrator's fingerprint and flags a worker that pinned a different one. Changing `DV_SIGNING_KEY` means resetting every worker's state.
 
+**Device host keys** follow the same model. A device's SSH host key is pinned at its first collected backup: the collector reports the key the device presented in its signed `collected` event, and the orchestrator stores it on the device if it has none, accepting it only from a collector of the device's own collection group. Every later `job.collect` task carries the pinned key in its signed payload, and the collector refuses any other key before authenticating (a permanent failure, reason `auth`). An administrator of the device's group forgets the key when the device is replaced or re-keyed. Collectors keep no known-hosts state of their own.
+
 **Limitations:**
 - Broker credentials are shared per role, so revocation is enforced cryptographically (signatures and keys), not at the broker. Per-worker broker users can be added later.
 - Celery remote control stays enabled because Flower needs it, so anyone holding broker credentials can still send Celery control commands (for example, to stop a worker consuming). They cannot forge assignments or tasks.
diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md
index f5e482a..dbfffa8 100644
--- a/docs/DEPLOYMENT.md
+++ b/docs/DEPLOYMENT.md
@@ -79,6 +79,16 @@ not exist, the bootstrap admin from `DV_ADMIN_USERNAME` / `DV_ADMIN_PASSWORD` as
 a member of it. Put a TLS-terminating proxy or load balancer in front of `web`,
 and set `DV_ALLOWED_HOSTS` and `DV_CSRF_TRUSTED_ORIGINS` to match.
 
+Set `DV_TRUSTED_PROXIES` to that proxy's address (or its network, comma separated
+for several). The `web` service records the address that connected to it, which
+is then the proxy rather than the user; with the proxy listed, the orchestrator
+takes the address the proxy reports instead. Without it every user appears to come
+from the proxy: sign-in lockout by address then locks everyone out together, an API
+token's allowed networks are compared with the proxy, and the audit log shows only
+the proxy. The proxy must append to (or replace) `X-Forwarded-For`, as nginx,
+HAProxy (`option forwardfor`), Traefik and cloud load balancers do. Never list an
+address that clients could connect from themselves.
+
 ### Users and access
 
 Sign-in uses a server-side session: an HttpOnly cookie (`Secure` unless
@@ -86,7 +96,8 @@ Sign-in uses a server-side session: an HttpOnly cookie (`Secure` unless
 orchestrator node shares it. Sessions end after `DV_SESSION_IDLE_MIN` without
 use and `DV_SESSION_MAX_HOURS` after sign-in. Repeated failed sign-ins lock out a
 username or client address for `DV_LOGIN_LOCKOUT_MIN`; the client address is the
-last `X-Forwarded-For` hop, which the `web` service adds.
+last `X-Forwarded-For` hop that is not one of `DV_TRUSTED_PROXIES` (with none set,
+the hop the `web` service adds).
 
 **API tokens** let scripts and monitoring call the API without a session: send
 `Authorization: Bearer dvt_...`. A token acts as its owner and never beyond their
@@ -199,6 +210,23 @@ log line matches. To automate approval: `manage.py approve_worker <name>
 - **Revoke** stops a worker consuming. **Forget** removes it; if it is still
   running it registers again as pending.
 
+### Device SSH host keys
+
+Collectors pin each device's SSH host key on first use. A device's first backup
+over SSH accepts the key the device presents; the orchestrator records it on the
+device (audited as `device.host_key_pin`, and noted in that backup's log) and sends
+it with every later backup. From then on a collector connects only if the device
+presents exactly that key, before any username or password is sent, so a host
+answering in the device's place cannot collect its credentials. A backup refused
+this way fails at once with the reason `auth` and is not retried.
+
+The device's page shows the pinned key's type and `SHA256:` fingerprint; compare it
+with the device's own (`show crypto key mypubkey` and the like) when the first
+backup is not made over a network you trust. When a device is replaced or its keys
+are regenerated, an administrator of its device group chooses **Forget host key**
+there (audited as `device.host_key_forget`), and the next backup pins the new key.
+Only a collector of the device's own collection group can pin a key.
+
 ### Storage targets
 
 Under **Admin Settings > Storage Targets**, create a target per place backups go:
@@ -722,6 +750,7 @@ secrets). The file takes precedence.
 | `DV_DEBUG` | `false` | Development mode; never in production |
 | `DV_ALLOWED_HOSTS` | none (`*` in debug) | Hostnames the UI is served on |
 | `DV_CSRF_TRUSTED_ORIGINS` | none | Full origins (`https://host`) behind HTTPS |
+| `DV_TRUSTED_PROXIES` | none | Addresses or networks of the proxies in front of `web`; the client address is the last `X-Forwarded-For` hop that is not one of them |
 | `DV_TIMEZONE` | `UTC` | Display timezone (storage is always UTC) |
 | `DV_LOG_LEVEL` | `INFO` | |
 | `DV_DB_ENGINE` | `sqlite` | `sqlite` or `postgres` |
diff --git a/frontend/src/api/endpoints.ts b/frontend/src/api/endpoints.ts
index 7c7887f..77f3184 100644
--- a/frontend/src/api/endpoints.ts
+++ b/frontend/src/api/endpoints.ts
@@ -164,6 +164,8 @@ export const devicesApi = {
   deactivate: async (id: Id, reason: string) =>
     (await http.post<DeviceDetail>(`/devices/${id}/deactivate/`, { reason })).data,
   reactivate: async (id: Id) => (await http.post<DeviceDetail>(`/devices/${id}/reactivate/`)).data,
+  /** Drop the pinned SSH host key (a replaced or re-keyed device); the next backup pins the new one. */
+  forgetHostKey: async (id: Id) => (await http.post<DeviceDetail>(`/devices/${id}/forget_host_key/`)).data,
   /** Global administrators: unpins and deletes every backup of an inactive device. `confirm` is its name. */
   purge: async (id: Id, confirm: string) =>
     (await http.post<DeviceDetail>(`/devices/${id}/purge/`, { confirm })).data,
diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts
index 75d0445..ec9e9db 100644
--- a/frontend/src/api/types.ts
+++ b/frontend/src/api/types.ts
@@ -225,6 +225,11 @@ interface DeviceBase {
   storage_target: string | null
   credential: Id | null
   schedule: Id | null
+  /** The SSH host key pinned at the device's first backup, which collectors then require; '' while none is. */
+  ssh_host_key_type: string
+  /** `SHA256:...`, as the device and `ssh-keygen -l` show it. */
+  ssh_host_key_fingerprint: string
+  ssh_host_key_pinned_at: IsoDateTime | null
   /** What the signed-in user holds on this device's group. */
   user_permissions: GrantFlag[]
   inactive_at: IsoDateTime | null
diff --git a/frontend/src/pages/EditDevicePage.vue b/frontend/src/pages/EditDevicePage.vue
index d82653d..c06a3c5 100644
--- a/frontend/src/pages/EditDevicePage.vue
+++ b/frontend/src/pages/EditDevicePage.vue
@@ -28,6 +28,33 @@
             {{ device.last_backup_time ? formatDateTime(device.last_backup_time) : 'Never' }}
           </q-badge>
         </div>
+        <div v-if="device.backup_method" class="col-12">
+          <div class="text-subtitle2">SSH host key</div>
+          <template v-if="device.ssh_host_key_fingerprint">
+            <div class="host-key">{{ device.ssh_host_key_type }} {{ device.ssh_host_key_fingerprint }}</div>
+            <div class="text-caption text-grey-7">
+              Pinned
+              {{ device.ssh_host_key_pinned_at ? formatDateTime(device.ssh_host_key_pinned_at) : '' }}
+              at the first backup. Backups fail if the device presents any other key.
+            </div>
+            <q-btn
+              v-if="isAdmin"
+              flat
+              dense
+              no-caps
+              color="negative"
+              icon="key_off"
+              label="Forget host key"
+              class="q-mt-xs"
+              :loading="forgetting"
+              @click="forgetHostKey"
+            />
+          </template>
+          <div v-else class="text-caption text-grey-7">
+            None pinned yet. The key the device presents at its first backup over SSH is pinned and required from
+            then on.
+          </div>
+        </div>
       </q-card-section>
     </q-card>
 
@@ -211,6 +238,7 @@
 </template>
 
 <script setup lang="ts">
+import { useQuasar } from 'quasar'
 import { computed, onMounted, ref } from 'vue'
 import { useRoute, useRouter } from 'vue-router'
 
@@ -252,6 +280,7 @@ const route = useRoute()
 const router = useRouter()
 const auth = useAuthStore()
 const notify = useNotify()
+const $q = useQuasar()
 
 const deviceId = computed(() => (route.name === 'device-edit' ? Number(route.params.id) : null))
 const isEdit = computed(() => deviceId.value !== null)
@@ -360,6 +389,31 @@ function toForm(detail: DeviceDetail): DeviceWrite {
   }
 }
 
+const forgetting = ref(false)
+
+function forgetHostKey(): void {
+  const current = device.value
+  if (!current) return
+  $q.dialog({
+    title: 'Forget the SSH host key?',
+    message:
+      `The next backup of ${current.name} will trust and pin whatever key the device at ` +
+      `${current.ip_address} presents. Do this only if the device was replaced or re-keyed.`,
+    cancel: true,
+    ok: { label: 'Forget host key', color: 'negative', flat: true },
+  }).onOk(async () => {
+    forgetting.value = true
+    try {
+      device.value = await devicesApi.forgetHostKey(current.id)
+      notify.success('Host key forgotten')
+    } catch (error) {
+      notify.error(error, 'Failed to forget the host key')
+    } finally {
+      forgetting.value = false
+    }
+  })
+}
+
 async function save(): Promise<void> {
   saving.value = true
   try {
@@ -419,4 +473,8 @@ onMounted(async () => {
 .device-form > * {
   max-width: 60ch;
 }
+.host-key {
+  font-family: monospace;
+  overflow-wrap: anywhere;
+}
 </style>
diff --git a/orchestrator/accounts/tests/test_auth.py b/orchestrator/accounts/tests/test_auth.py
index 756743f..5bc1388 100644
--- a/orchestrator/accounts/tests/test_auth.py
+++ b/orchestrator/accounts/tests/test_auth.py
@@ -137,6 +137,51 @@ def test_the_client_address_comes_from_the_proxy_hop(alice):
     assert AuditLog.objects.get(action='auth.login').ip == '203.0.113.5'
 
 
+def _address(settings, forwarded, proxies=()):
+    import ipaddress
+
+    from django.test import RequestFactory
+
+    from audit.log import client_ip
+
+    settings.DEVICEVAULT_TRUSTED_PROXIES = tuple(ipaddress.ip_network(proxy) for proxy in proxies)
+    return client_ip(RequestFactory().get('/', REMOTE_ADDR='172.18.0.5', HTTP_X_FORWARDED_FOR=forwarded))
+
+
+@pytest.mark.parametrize(
+    ('forwarded', 'proxies', 'expected'),
+    [
+        # A TLS proxy in front of nginx: the client is the hop the proxy added, not the proxy.
+        ('198.51.100.7, 10.0.0.2', ['10.0.0.0/24'], '198.51.100.7'),
+        # What the client sent itself (left of the proxy's hop) is never believed.
+        ('192.0.2.99, 198.51.100.7, 10.0.0.2', ['10.0.0.0/24'], '198.51.100.7'),
+        # Two proxies in a row, IPv4 and IPv6.
+        ('198.51.100.7, 2001:db8::1, 10.0.0.2', ['10.0.0.2/32', '2001:db8::/64'], '198.51.100.7'),
+        ('198.51.100.7, ::ffff:10.0.0.2', ['10.0.0.0/24'], '198.51.100.7'),
+        # A request from the proxy itself (a health check).
+        ('10.0.0.2', ['10.0.0.0/24'], '10.0.0.2'),
+        # A client pretending to be a proxy gains nothing: its own hop is still what the proxy wrote.
+        ('10.0.0.9, 198.51.100.7, 10.0.0.2', ['10.0.0.0/24'], '198.51.100.7'),
+        ('junk, 10.0.0.2', ['10.0.0.0/24'], None),
+        ('198.51.100.7, 10.0.0.2', [], '10.0.0.2'),
+    ],
+)
+def test_trusted_proxies_are_skipped_to_find_the_client(settings, forwarded, proxies, expected):
+    assert _address(settings, forwarded, proxies) == expected
+
+
+def test_behind_a_trusted_proxy_one_client_cannot_lock_everyone_out(alice, settings):
+    import ipaddress
+
+    settings.DEVICEVAULT_LOGIN_MAX_FAILURES = 2
+    settings.DEVICEVAULT_TRUSTED_PROXIES = (ipaddress.ip_network('10.0.0.2/32'),)
+    for n in range(3):
+        sign_in(APIClient(HTTP_X_FORWARDED_FOR='198.51.100.7, 10.0.0.2'), username=f'guess-{n}', password='nope')
+
+    assert sign_in(APIClient(HTTP_X_FORWARDED_FOR='198.51.100.7, 10.0.0.2')).status_code == 429
+    assert sign_in(APIClient(HTTP_X_FORWARDED_FOR='203.0.113.5, 10.0.0.2')).status_code == 200
+
+
 def test_disabled_users_cannot_sign_in_and_lose_their_session(alice, admin_client):
     client = APIClient()
     sign_in(client)
diff --git a/orchestrator/api/serializers.py b/orchestrator/api/serializers.py
index bcdccfc..073aa7e 100644
--- a/orchestrator/api/serializers.py
+++ b/orchestrator/api/serializers.py
@@ -275,6 +275,9 @@ class Meta:
             'storage_target',
             'schedule',
             'credential',
+            'ssh_host_key_type',
+            'ssh_host_key_fingerprint',
+            'ssh_host_key_pinned_at',
             'user_permissions',
             *LIFECYCLE_FIELDS,
         ]
@@ -353,6 +356,9 @@ class Meta:
             'storage_target',
             'schedule',
             'credential',
+            'ssh_host_key_type',
+            'ssh_host_key_fingerprint',
+            'ssh_host_key_pinned_at',
             'user_permissions',
             *LIFECYCLE_FIELDS,
         ]
diff --git a/orchestrator/api/tests/test_devices.py b/orchestrator/api/tests/test_devices.py
index f251d9b..0b880a3 100644
--- a/orchestrator/api/tests/test_devices.py
+++ b/orchestrator/api/tests/test_devices.py
@@ -42,3 +42,43 @@ def test_version_needs_no_authentication(db):
 
     assert resp.status_code == 200
     assert resp.json() == {'version': version()}
+
+
+HOST_KEY = 'ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIJ/5z9AzdT8Dm9DM5cwhVsGr+7YDGjr3iqkJqscQWMlL'
+
+
+def test_a_device_shows_its_pinned_host_key_as_a_fingerprint(device, admin_client):
+    assert admin_client.get(f'/api/devices/{device.pk}/').json()['ssh_host_key_fingerprint'] == ''
+    Device.objects.filter(pk=device.pk).update(ssh_host_key=HOST_KEY)
+
+    body = admin_client.get(f'/api/devices/{device.pk}/').json()
+
+    assert body['ssh_host_key_type'] == 'ssh-ed25519'
+    assert body['ssh_host_key_fingerprint'] == 'SHA256:kuehv88oMZA/RgHRswCgnfGgnwqVeQ91l54enX5Lzec'
+    assert 'ssh_host_key' not in body
+
+
+def test_the_host_key_cannot_be_set_through_the_api(device, admin_client):
+    reply = admin_client.patch(f'/api/devices/{device.pk}/', {'ssh_host_key': HOST_KEY}, format='json')
+
+    assert reply.status_code == 200
+    assert Device.objects.get(pk=device.pk).ssh_host_key == ''
+
+
+def test_group_admins_forget_a_host_key_and_it_is_audited(device, make_user, client_for):
+    from audit.models import AuditLog
+
+    Device.objects.filter(pk=device.pk).update(ssh_host_key=HOST_KEY)
+    viewer = client_for(make_user('viewer', grants={device.device_group: ['view', 'backup_now']}))
+    admin = client_for(make_user('groupadmin', grants={device.device_group: ['admin']}))
+
+    assert viewer.post(f'/api/devices/{device.pk}/forget_host_key/').status_code == 403
+    assert Device.objects.get(pk=device.pk).ssh_host_key == HOST_KEY
+
+    reply = admin.post(f'/api/devices/{device.pk}/forget_host_key/')
+
+    assert reply.status_code == 200 and reply.json()['ssh_host_key_fingerprint'] == ''
+    device.refresh_from_db()
+    assert device.ssh_host_key == '' and device.ssh_host_key_pinned_at is None
+    entry = AuditLog.objects.get(action='device.host_key_forget')
+    assert entry.actor_name == 'groupadmin' and entry.details['key_type'] == 'ssh-ed25519'
diff --git a/orchestrator/api/views.py b/orchestrator/api/views.py
index bf434c4..c3a477a 100644
--- a/orchestrator/api/views.py
+++ b/orchestrator/api/views.py
@@ -135,6 +135,7 @@ class DeviceViewSet(viewsets.ModelViewSet):
     ``POST {id}/backup_now/`` needs ``backup_now``. Retiring a device
     (``deactivate/``, ``reactivate/``, and ``DELETE`` once it is deletable) needs
     ``admin``; ``purge/`` is for global administrators (devices.lifecycle).
+    ``POST {id}/forget_host_key/`` (``admin``) drops the SSH host key pinned at the device's first backup.
     ``?lifecycle=active|inactive|deletable`` filters the list.
     """
 
@@ -236,6 +237,27 @@ def reactivate(self, request, pk=None):
         )
         return self._detail()
 
+    @decorators.action(detail=True, methods=['post'], permission_classes=[HasDeviceGroupPermission('admin')])
+    def forget_host_key(self, request, pk=None):
+        """Drop the pinned SSH host key (the device was replaced or re-keyed): its next backup pins the new one."""
+        device = self.get_object()
+        if device.ssh_host_key:
+            details = {
+                'device': device.pk,
+                'device_name': device.name,
+                'key_type': device.ssh_host_key_type,
+                'fingerprint': device.ssh_host_key_fingerprint,
+            }
+            Device.objects.filter(pk=device.pk).update(ssh_host_key='', ssh_host_key_pinned_at=None)
+            record(
+                request,
+                'device.host_key_forget',
+                resource=str(device.pk),
+                device_group=device.device_group_id,
+                details=details,
+            )
+        return self._detail()
+
     @decorators.action(detail=True, methods=['post'], permission_classes=[IsGlobalAdmin])
     def purge(self, request, pk=None):
         """End an inactive device's hold now: unpin and delete all of its backups."""
diff --git a/orchestrator/audit/log.py b/orchestrator/audit/log.py
index b2be3a7..694ba98 100644
--- a/orchestrator/audit/log.py
+++ b/orchestrator/audit/log.py
@@ -3,19 +3,34 @@
 import ipaddress
 import logging
 
+from django.conf import settings
+
 from audit.models import AuditLog
 
 logger = logging.getLogger('devicevault.audit')
 
 
 def client_ip(request) -> str | None:
-    """The client's address: the last X-Forwarded-For hop (added by our nginx), else the peer."""
+    """The client's address, from X-Forwarded-For (our nginx appends the peer it saw), else the peer.
+
+    Read from the right, skipping ``DV_TRUSTED_PROXIES`` (a TLS proxy or load balancer in front of
+    nginx): the first hop that is not one of them is the client, since everything to its left is
+    whatever the client chose to send. With no trusted proxies that is the hop nginx added.
+    """
     forwarded = request.META.get('HTTP_X_FORWARDED_FOR', '')
-    address = forwarded.split(',')[-1].strip() if forwarded else request.META.get('REMOTE_ADDR', '')
-    try:
-        return str(ipaddress.ip_address(address))
-    except ValueError:
-        return None
+    hops = [hop.strip() for hop in forwarded.split(',')] if forwarded else [request.META.get('REMOTE_ADDR', '')]
+    address = None
+    for hop in reversed(hops):
+        try:
+            address = ipaddress.ip_address(hop)
+        except ValueError:
+            return None  # a hop a trusted proxy would never write: nothing left of here can be believed
+        if address.version == 6 and address.ipv4_mapped:
+            address = address.ipv4_mapped
+        if not any(address in network for network in settings.DEVICEVAULT_TRUSTED_PROXIES):
+            break
+    # Every hop a trusted proxy: the request came from the outermost one itself.
+    return str(address) if address is not None else None
 
 
 def _token_prefix(request) -> str:
diff --git a/orchestrator/backups/jobs.py b/orchestrator/backups/jobs.py
index 7806a2c..8a7625f 100644
--- a/orchestrator/backups/jobs.py
+++ b/orchestrator/backups/jobs.py
@@ -28,6 +28,7 @@
     EVENT_STORED,
     STAGE_STORE,
     failure_reason,
+    host_key_fingerprint,
 )
 from devicevault_protocol.names import COLLECT_TASK, STORE_TASK, collection_queue_name
 from devicevault_protocol.registration import ROLE_COLLECTOR, ROLE_STORAGE, STATUS_APPROVED
@@ -127,7 +128,13 @@ def collect_payload(job: BackupJob) -> dict[str, Any]:
     store = store_spec(job)
     collect = {
         'job_id': str(job.id),
-        'device': {'id': device.pk, 'name': device.name, 'address': device.ip_address},
+        'device': {
+            'id': device.pk,
+            'name': device.name,
+            'address': device.ip_address,
+            # Pinned at the device's first SSH backup; the collector refuses any other key.
+            'host_key': device.ssh_host_key or None,
+        },
         'driver': job.driver,
         'driver_params': device.driver_params or {},
         'timeout': settings.DEVICEVAULT_COLLECT_TIMEOUT_SEC,
@@ -221,6 +228,7 @@ def _apply(job: BackupJob, payload: dict[str, Any], worker: Worker) -> None:
         job.collect_duration_ms = int(payload.get('duration_ms') or 0)
         job.collected_at = now
         job.error = ''
+        _pin_host_key(job, payload, worker)
         if job.state in (BackupJob.QUEUED, BackupJob.COLLECTING, BackupJob.TIMED_OUT):
             job.state = BackupJob.COLLECTED
             job.failure_reason = None  # a timed-out job that turned up after all
@@ -238,6 +246,56 @@ def _apply(job: BackupJob, payload: dict[str, Any], worker: Worker) -> None:
         job.finished_at = now
 
 
+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.
+    """
+    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:
+        return
+    fingerprint = host_key_fingerprint(key)
+    if fingerprint is None:
+        logger.warning('host_key_ignored', extra={'job_id': str(job.id), 'reason': 'not a host key'})
+        return
+    if device.ssh_host_key:
+        if device.ssh_host_key != key:
+            # 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()
+    )
+    if not pinned:
+        return  # another job pinned one meanwhile
+    device.ssh_host_key = key
+    message = f'pinned SSH host key {key.split(" ")[0]} {fingerprint} (first connection)'
+    job.log = (job.log + [_log_entry('WARNING', message, source='orchestrator')])[-MAX_LOG_ENTRIES:]
+    record(
+        None,
+        'device.host_key_pin',
+        resource=str(device.pk),
+        device_group=device.device_group_id,
+        details={
+            'device': device.pk,
+            'device_name': device.name,
+            'key_type': key.split(' ')[0],
+            'fingerprint': fingerprint,
+            'collector': worker.name,
+            'job': str(job.id),
+        },
+        actor_name=worker.name,
+    )
+
+
 def _reported_reason(payload: dict[str, Any]) -> str:
     """The failure reason for a ``failed`` event: a storage-step failure is ``storage`` (or ``integrity``)."""
     reason = failure_reason(payload.get('reason'))
@@ -266,8 +324,8 @@ def _apply_deletion(job: BackupJob, payload: dict[str, Any], worker: Worker) ->
     job.log = (job.log + entries)[-MAX_LOG_ENTRIES:]
 
 
-def _log_entry(severity: str, message: str) -> dict[str, str]:
-    return {'source': 'retention', 'timestamp': timezone.now().isoformat(), 'severity': severity, 'message': message}
+def _log_entry(severity: str, message: str, *, source: str = 'retention') -> dict[str, str]:
+    return {'source': source, 'timestamp': timezone.now().isoformat(), 'severity': severity, 'message': message}
 
 
 def _finish_if_complete(job: BackupJob, now) -> None:
diff --git a/orchestrator/backups/tests/test_jobs.py b/orchestrator/backups/tests/test_jobs.py
index 4497bdf..8f5b924 100644
--- a/orchestrator/backups/tests/test_jobs.py
+++ b/orchestrator/backups/tests/test_jobs.py
@@ -30,7 +30,7 @@ def test_a_job_is_dispatched_as_signed_tasks_for_collector_and_storage(
     assert (name, queue) == ('job.collect', 'collector.group.dmz')
     task = verify_task(envelope, ORCHESTRATOR())
     assert task['job_id'] == str(job.id)
-    assert task['device'] == {'id': setup['device'].id, 'name': 'Edge 01', 'address': '192.0.2.10'}
+    assert task['device'] == {'id': setup['device'].id, 'name': 'Edge 01', 'address': '192.0.2.10', 'host_key': None}
     assert task['store']['queue'] == f'storage.target.{setup["target"].id}'
     store = verify_task(task['store']['envelope'], ORCHESTRATOR())
     assert store['config'] == {'path': 'fs'} and store['target']['driver'] == 'filesystem'
@@ -231,3 +231,77 @@ def test_failure_reason_is_read_only_in_the_api():
 
     assert 'failure_reason' in BackupJobSerializer.Meta.fields
     assert 'failure_reason' in BackupJobSerializer.Meta.read_only_fields
+
+
+# --- SSH host keys (pinned on first use) ----------------------------------------------------------
+
+HOST_KEY = 'ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIJ/5z9AzdT8Dm9DM5cwhVsGr+7YDGjr3iqkJqscQWMlL'
+OTHER_KEY = 'ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAILDOn0OvFfw6p/WXBHAsaPNJxLtOnAjNRemDF7b3iKed'
+
+
+def collected_with_key(setup, job, key, keys=None):
+    event = job_event(keys or setup['collector'], job, EVENT_COLLECTED, sha256=SHA, size=10, host_key=key)
+    return handle_job_event(event)
+
+
+def test_the_first_collected_backup_pins_the_devices_host_key(setup, broker):
+    from audit.models import AuditLog
+
+    job, _ = create_job(setup['device'])
+
+    job = collected_with_key(setup, job, HOST_KEY)
+
+    device = setup['device']
+    device.refresh_from_db()
+    assert device.ssh_host_key == HOST_KEY and device.ssh_host_key_pinned_at is not None
+    assert device.ssh_host_key_fingerprint.startswith('SHA256:') and device.ssh_host_key_type == 'ssh-ed25519'
+    assert device.ssh_host_key_fingerprint in job.log[-1]['message']
+    entry = AuditLog.objects.get(action='device.host_key_pin')
+    assert entry.actor_name == 'collector-1' and entry.details['fingerprint'] == device.ssh_host_key_fingerprint
+
+
+def test_the_pinned_host_key_travels_in_the_signed_collect_task(setup, broker, django_capture_on_commit_callbacks):
+    setup['device'].ssh_host_key = HOST_KEY
+    setup['device'].save()
+
+    queue_job(setup, django_capture_on_commit_callbacks)
+
+    [(_, [envelope], _)] = broker.sent
+    assert verify_task(envelope, ORCHESTRATOR())['device']['host_key'] == HOST_KEY
+
+
+def test_a_pinned_host_key_is_never_replaced_by_an_event(setup, broker):
+    setup['device'].ssh_host_key = HOST_KEY
+    setup['device'].save()
+    job, _ = create_job(setup['device'])
+
+    collected_with_key(setup, job, OTHER_KEY)
+
+    setup['device'].refresh_from_db()
+    assert setup['device'].ssh_host_key == HOST_KEY
+
+
+def test_only_the_devices_own_collectors_can_pin_its_host_key(setup, broker, worker_keys):
+    from core import workers
+    from devices.models import CollectionGroup
+
+    elsewhere = worker_keys('collector-2')
+    other_group = CollectionGroup.objects.create(name='Lab', rabbitmq_queue_id='lab')
+    workers.approve(workers.handle_event(elsewhere.event()), role='collector', collection_groups=[other_group])
+    job, _ = create_job(setup['device'])
+
+    collected_with_key(setup, job, HOST_KEY, keys=elsewhere)
+    collected_with_key(setup, job, HOST_KEY, keys=setup['storage'])
+
+    setup['device'].refresh_from_db()
+    assert setup['device'].ssh_host_key == ''
+
+
+@pytest.mark.parametrize('key', ['ssh-ed25519', 'ssh-ed25519 not base64!', 'ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI', 7])
+def test_malformed_host_keys_are_not_pinned(setup, broker, key):
+    job, _ = create_job(setup['device'])
+
+    assert collected_with_key(setup, job, key).state == BackupJob.COLLECTED
+
+    setup['device'].refresh_from_db()
+    assert setup['device'].ssh_host_key == ''
diff --git a/orchestrator/devices/migrations/0008_device_ssh_host_key.py b/orchestrator/devices/migrations/0008_device_ssh_host_key.py
new file mode 100644
index 0000000..3dc1c73
--- /dev/null
+++ b/orchestrator/devices/migrations/0008_device_ssh_host_key.py
@@ -0,0 +1,23 @@
+# Generated by Django 6.1.2 on 2026-11-15 04:13
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('devices', '0007_device_created_at'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='device',
+            name='ssh_host_key',
+            field=models.TextField(blank=True, help_text='Pinned SSH host key, as "<type> <base64>"'),
+        ),
+        migrations.AddField(
+            model_name='device',
+            name='ssh_host_key_pinned_at',
+            field=models.DateTimeField(blank=True, help_text='When the host key was pinned', null=True),
+        ),
+    ]
diff --git a/orchestrator/devices/models.py b/orchestrator/devices/models.py
index 8be70ef..7954036 100644
--- a/orchestrator/devices/models.py
+++ b/orchestrator/devices/models.py
@@ -222,6 +222,10 @@ class Device(models.Model):
         blank=True,
         help_text='SSH/Telnet credentials for device access',
     )
+    # Pinned on first use: the key the device presented at its first SSH backup, which collectors
+    # then require (backups.jobs). Cleared with the forget_host_key action when a device is replaced.
+    ssh_host_key = models.TextField(blank=True, help_text='Pinned SSH host key, as "<type> <base64>"')
+    ssh_host_key_pinned_at = models.DateTimeField(null=True, blank=True, help_text='When the host key was pinned')
     inactive_at = models.DateTimeField(null=True, blank=True, help_text='When the device was marked inactive')
     inactive_by = models.ForeignKey(User, null=True, blank=True, on_delete=models.SET_NULL, related_name='+')
     inactive_reason = models.CharField(max_length=200, blank=True, help_text='Why the device was retired')
@@ -247,6 +251,17 @@ def __str__(self):
     def is_active(self) -> bool:
         return self.inactive_at is None
 
+    @property
+    def ssh_host_key_type(self) -> str:
+        return self.ssh_host_key.split(' ')[0] if self.ssh_host_key else ''
+
+    @property
+    def ssh_host_key_fingerprint(self) -> str:
+        """``SHA256:...`` of the pinned host key, to compare with what the device shows ('' when none is pinned)."""
+        from devicevault_protocol.jobs import host_key_fingerprint
+
+        return host_key_fingerprint(self.ssh_host_key) or ''
+
 
 # ===== Device groups =====
 
diff --git a/orchestrator/devicevault/settings.py b/orchestrator/devicevault/settings.py
index 66be930..b06180e 100644
--- a/orchestrator/devicevault/settings.py
+++ b/orchestrator/devicevault/settings.py
@@ -17,6 +17,7 @@
 # All configuration comes from DV_* environment variables; each also accepts
 # <NAME>_FILE for Docker secrets. See deploy/orchestrator/orchestrator.env.example.
 
+import ipaddress
 from pathlib import Path
 
 from devicevault_protocol.bus import BusSettings
@@ -134,6 +135,15 @@
 SECURE_PROXY_SSL_HEADER = ('HTTP_X_FORWARDED_PROTO', 'https')
 USE_X_FORWARDED_HOST = True
 
+# Proxies in front of the web tier (addresses or networks). The client address is the last
+# X-Forwarded-For hop that is not one of them; with none, it is the hop the web tier added.
+try:
+    DEVICEVAULT_TRUSTED_PROXIES = tuple(
+        ipaddress.ip_network(entry, strict=False) for entry in env_list('DV_TRUSTED_PROXIES')
+    )
+except ValueError as exc:
+    raise ConfigError(f'DV_TRUSTED_PROXIES must list addresses or networks: {exc}') from exc
+
 # Server-side sessions in the database (shared by every orchestrator node), used
 # by the UI and the API alike. See accounts.sessions for the timeouts.
 SESSION_ENGINE = 'django.contrib.sessions.backends.db'
diff --git a/protocol/src/devicevault_protocol/jobs.py b/protocol/src/devicevault_protocol/jobs.py
index b91cfa0..cf91caa 100644
--- a/protocol/src/devicevault_protocol/jobs.py
+++ b/protocol/src/devicevault_protocol/jobs.py
@@ -22,7 +22,7 @@
      "worker_id": "<uuid>", "attempt": 1, "log": [<log entry>, ...], ...}
 
 ``collected`` adds ``sha256``, ``size``, ``is_binary``, ``encrypted``,
-``duration_ms``; ``stored`` adds ``sha256``, ``size``, ``storage_ref``,
+``duration_ms`` and, from SSH drivers, ``host_key`` (see below); ``stored`` adds ``sha256``, ``size``, ``storage_ref``,
 ``duration_ms``; ``retrying`` and ``failed`` add ``error`` and ``stage``
 ("collect" or "store"); ``failed`` also adds ``reason``, one of
 ``FAILURE_REASONS``. The orchestrator stores it as ``BackupJob.failure_reason``
@@ -31,6 +31,13 @@
 The orchestrator only accepts a ``stored`` event whose hash matches the one
 the collector reported.
 
+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.
+
 Retention: the orchestrator sends ``storage.delete`` (signed) to a storage
 target's queue with ``items`` (``[{"job_id", "storage_ref"}, ...]``); the
 storage worker reports ``deleted`` or ``delete_failed`` (adds ``error``) per
@@ -40,8 +47,28 @@
 Log entries: ``{"source", "timestamp", "severity", "message"}``.
 """
 
+import base64
+import binascii
+import hashlib
+import re
+
 JOB_EVENT_TASK = 'orchestrator.job_event'
 
+# An SSH host key as it is pinned and sent: ``<key type> <base64 key>``.
+HOST_KEY_PATTERN = re.compile(r'[a-z0-9][a-z0-9@._-]{2,63} [A-Za-z0-9+/]{16,4096}={0,2}')
+
+
+def host_key_fingerprint(value: object) -> str | None:
+    """``SHA256:...`` of a host key, as ``ssh-keygen -l`` and devices show it; None if it is not a host key."""
+    if not isinstance(value, str) or HOST_KEY_PATTERN.fullmatch(value) is None:
+        return None
+    try:
+        blob = base64.b64decode(value.split(' ')[1], validate=True)
+    except binascii.Error, ValueError:
+        return None
+    return 'SHA256:' + base64.b64encode(hashlib.sha256(blob).digest()).decode('ascii').rstrip('=')
+
+
 EVENT_STARTED = 'started'
 EVENT_RETRYING = 'retrying'
 EVENT_COLLECTED = 'collected'
diff --git a/worker/src/devicevault_worker/collect.py b/worker/src/devicevault_worker/collect.py
index 3e70aea..35d6094 100644
--- a/worker/src/devicevault_worker/collect.py
+++ b/worker/src/devicevault_worker/collect.py
@@ -119,6 +119,7 @@ def collect(self, envelope: dict) -> None:
             encrypted=encrypted,
             duration_ms=int((time.monotonic() - started) * 1000),
             log=log.entries,
+            **({'host_key': log.host_key} if log.host_key else {}),
         )
         store = job['store']
         app.send_task(STORE_TASK, args=[store['envelope'], blob], queue=store['queue'])
@@ -149,7 +150,7 @@ def _collect(job: dict[str, Any], log: CollectionLog) -> tuple[bytes, bool]:
     except ValueError as exc:
         raise PermanentError(f'invalid driver parameters: {exc}') from exc
     device = job['device']
-    target = Target(address=device['address'], name=device.get('name', ''))
+    target = Target(address=device['address'], name=device.get('name', ''), host_key=device.get('host_key') or '')
     credentials = secret(job, 'credentials') or {}
     content = driver_cls().collect(target, credentials, params, timeout=int(job.get('timeout', 120)), log=log)
     if isinstance(content, str):
diff --git a/worker/src/devicevault_worker/drivers/base.py b/worker/src/devicevault_worker/drivers/base.py
index d1720ea..6f83dae 100644
--- a/worker/src/devicevault_worker/drivers/base.py
+++ b/worker/src/devicevault_worker/drivers/base.py
@@ -57,6 +57,9 @@ class Target:
     address: str
     name: str = ''
     port: int | None = None
+    # The SSH host key pinned for the device (``<type> <base64>``), or '' before its first backup.
+    # A driver that connects over SSH must refuse any other key (``drivers.ssh``).
+    host_key: str = ''
 
 
 @dataclass
@@ -65,6 +68,9 @@ class CollectionLog:
 
     source: str
     entries: list[dict[str, str]] = field(default_factory=list)
+    # The SSH host key the device presented (``<type> <base64>``), set by drivers that connect
+    # over SSH. It is reported with the collected backup, and the orchestrator pins it on first use.
+    host_key: str = ''
 
     def add(self, severity: str, message: str) -> None:
         self.entries.append(
diff --git a/worker/src/devicevault_worker/drivers/ssh.py b/worker/src/devicevault_worker/drivers/ssh.py
index cb4ee93..bbaa04b 100644
--- a/worker/src/devicevault_worker/drivers/ssh.py
+++ b/worker/src/devicevault_worker/drivers/ssh.py
@@ -6,14 +6,24 @@
 
 Credentials: ``username`` and ``password``, plus ``enable_secret`` for
 platforms with a privileged mode.
+
+Host keys are pinned on first use: with ``target.host_key`` set, the device
+must present exactly that key, or the collection fails before any credential
+is sent. Without one (the device's first backup) any key is accepted and
+reported in ``log.host_key`` for the orchestrator to pin.
 """
 
+import contextlib
+import os
 import re
+import tempfile
+from collections.abc import Iterator
 from typing import Any, ClassVar
 
 from netmiko import ConnectHandler, NetmikoAuthenticationException, NetmikoTimeoutException
 from netmiko.exceptions import ReadTimeout
-from paramiko.ssh_exception import SSHException
+from paramiko.hostkeys import InvalidHostKey
+from paramiko.ssh_exception import BadHostKeyException, SSHException
 from pydantic import Field
 
 from devicevault_protocol.jobs import FAILURE_AUTH, FAILURE_COMMAND, FAILURE_CONNECTION, FAILURE_TIMEOUT
@@ -24,6 +34,32 @@
 REJECTION_LINES = 3
 
 
+def known_hosts_name(address: str, port: int) -> str:
+    """How known_hosts (and Paramiko's lookup) names a server: bare on port 22, else ``[address]:port``."""
+    return address if port == 22 else f'[{address}]:{port}'
+
+
+@contextlib.contextmanager
+def pinned_known_hosts(address: str, port: int, host_key: str) -> Iterator[str]:
+    """A temporary known_hosts file holding only the device's pinned key."""
+    handle, path = tempfile.mkstemp(prefix='known-hosts-')
+    try:
+        with os.fdopen(handle, 'w', encoding='ascii') as out:
+            out.write(f'{known_hosts_name(address, port)} {host_key}\n')
+        yield path
+    finally:
+        os.unlink(path)
+
+
+def presented_host_key(connection) -> str:
+    """The host key of an open Netmiko connection as ``<type> <base64>``, or '' if it cannot be read."""
+    try:
+        key = connection.remote_conn_pre.get_transport().get_remote_server_key()
+        return f'{key.get_name()} {key.get_base64()}'
+    except AttributeError, SSHException:
+        return ''
+
+
 class SshParams(DriverParams):
     port: int = Field(22, ge=1, le=65535, description='SSH port')
     strip_volatile: bool = Field(
@@ -61,8 +97,8 @@ def connect_args(self, target: Target, credentials: dict[str, Any], params: SshP
             'conn_timeout': timeout,
             'auth_timeout': timeout,
             'banner_timeout': timeout,
-            # TODO(host keys): pin device host keys instead of accepting any.
-            'ssh_strict': False,
+            # Host keys are pinned per device (see ``collect``): never the worker's own known_hosts.
+            'ssh_strict': bool(target.host_key),
             'system_host_keys': False,
             'use_keys': False,
             'allow_agent': False,
@@ -82,9 +118,17 @@ def collect(self, target, credentials, params, *, timeout: int, log: CollectionL
         outputs = []
         connected = False
         try:
-            with ConnectHandler(**args) as connection:
+            with contextlib.ExitStack() as stack:
+                if target.host_key:
+                    # Paramiko then asks for the pinned key's type and refuses any other key.
+                    known_hosts = stack.enter_context(pinned_known_hosts(target.address, params.port, target.host_key))
+                    args.update(alt_host_keys=True, alt_key_file=known_hosts)
+                connection = stack.enter_context(ConnectHandler(**args))
                 connected = True
+                log.host_key = presented_host_key(connection)
                 log.info(f'connected to {where} as {args["username"]}')
+                if not target.host_key and log.host_key:
+                    log.warning(f'first connection: accepted host key {log.host_key.split()[0]}, to be pinned')
                 if args['secret']:
                     try:
                         connection.enable()
@@ -100,8 +144,33 @@ def collect(self, target, credentials, params, *, timeout: int, log: CollectionL
             raise PermanentError(
                 f'authentication failed for {args["username"]}@{target.address}', reason=FAILURE_AUTH
             ) from exc
+        except InvalidHostKey as exc:
+            raise PermanentError(
+                f'the host key pinned for {where} is unusable; forget it and back the device up again',
+                reason=FAILURE_AUTH,
+            ) from exc
         except (TimeoutError, NetmikoTimeoutException, ReadTimeout, SSHException, OSError, EOFError) as exc:
             if connected:  # the session was up: the device stopped answering
                 raise TransientError(f'{where} stopped responding: {exc}', reason=FAILURE_TIMEOUT) from exc
+            if target.host_key and _host_key_refused(exc):
+                raise PermanentError(
+                    f'{where} did not present its pinned SSH host key ({target.host_key.split()[0]}), so nothing was '
+                    'sent to it. If the device was replaced or re-keyed, forget its host key and back it up again',
+                    reason=FAILURE_AUTH,
+                ) from exc
             raise TransientError(f'connection to {where} failed: {exc}', reason=FAILURE_CONNECTION) from exc
         return self.clean('\n'.join(outputs), params).encode('utf-8')
+
+
+def _host_key_refused(exc: BaseException | None) -> bool:
+    """Whether Paramiko refused the server's host key: another key, or (RejectPolicy) one of a type not pinned.
+
+    Netmiko re-raises Paramiko's error as its own timeout exception, so the original is found in the chain.
+    """
+    while exc is not None:
+        if isinstance(exc, BadHostKeyException):
+            return True
+        if isinstance(exc, SSHException) and 'not found in known_hosts' in str(exc):
+            return True
+        exc = exc.__cause__ or exc.__context__
+    return False
diff --git a/worker/tests/test_drivers.py b/worker/tests/test_drivers.py
index b5db207..7532802 100644
--- a/worker/tests/test_drivers.py
+++ b/worker/tests/test_drivers.py
@@ -1,10 +1,11 @@
+import os
 import re
 import socket
 
 import pytest
 from netmiko import NetmikoAuthenticationException, NetmikoTimeoutException
 from netmiko.exceptions import ReadTimeout
-from paramiko.ssh_exception import SSHException
+from paramiko.ssh_exception import BadHostKeyException, SSHException
 
 from devicevault_worker.drivers import get_driver
 from devicevault_worker.drivers.base import CollectionLog, PermanentError, Target, TransientError
@@ -35,6 +36,11 @@ class FakeConnection:
 
     def __init__(self, outputs, error=None, **kwargs):
         self.kwargs = kwargs
+        # The pinned host key's known_hosts file only exists while the connection is being made.
+        self.known_hosts = None
+        if kwargs.get('alt_key_file'):
+            with open(kwargs['alt_key_file']) as known_hosts:
+                self.known_hosts = known_hosts.read()
         self.outputs = outputs
         self.error = error
         self.commands = []
@@ -259,3 +265,85 @@ def test_mikrotik_secret_lines():
     assert first_secrets('mikrotik_ssh_export', 'set enabled=yes ipsec-secret=l2tp use-ipsec=yes') == [('l2tp', None)]
     assert first_secrets('mikrotik_ssh_export', 'add private-key="abc=" public-key="def="') == [('"abc="', None)]
     assert get_driver('noop').describe()['secret_lines'] == []
+
+
+# --- host keys (pinned on first use) ------------------------------------------------------------
+
+HOST_KEY = 'ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIJ/5z9AzdT8Dm9DM5cwhVsGr+7YDGjr3iqkJqscQWMlL'
+
+
+class FakeHostKey:
+    def get_name(self):
+        return HOST_KEY.split()[0]
+
+    def get_base64(self):
+        return HOST_KEY.split()[1]
+
+
+class FakeSshClient:
+    def get_transport(self):
+        return self
+
+    def get_remote_server_key(self):
+        return FakeHostKey()
+
+
+def collect(netmiko_made, host_key='', **params):
+    driver = get_driver('cisco_ios')
+    log = CollectionLog(source='cisco_ios')
+    target = Target(address='192.0.2.1', name='edge-01', host_key=host_key)
+    driver().collect(target, CREDS, driver.Params.model_validate(params), timeout=10, log=log)
+    return netmiko_made[0], log
+
+
+def test_a_first_connection_accepts_the_host_key_and_reports_it(netmiko, monkeypatch):
+    monkeypatch.setattr(FakeConnection, 'remote_conn_pre', FakeSshClient(), raising=False)
+
+    connection, log = collect(netmiko({'show running-config': CISCO_OUTPUT}))
+
+    assert connection.kwargs['ssh_strict'] is False and 'alt_key_file' not in connection.kwargs
+    assert connection.kwargs['system_host_keys'] is False
+    assert log.host_key == HOST_KEY
+    assert any('to be pinned' in entry['message'] for entry in log.entries)
+
+
+@pytest.mark.parametrize(('port', 'name'), [(22, '192.0.2.1'), (2222, '[192.0.2.1]:2222')])
+def test_a_pinned_host_key_is_the_only_one_accepted(netmiko, port, name):
+    """Strict checking against a known_hosts file holding just the pinned key, removed afterwards."""
+    connection, _ = collect(netmiko({'show running-config': CISCO_OUTPUT}), host_key=HOST_KEY, port=port)
+
+    assert connection.kwargs['ssh_strict'] is True and connection.kwargs['alt_host_keys'] is True
+    assert connection.known_hosts == f'{name} {HOST_KEY}\n'
+    assert not os.path.exists(connection.kwargs['alt_key_file'])
+
+
+def _as_netmiko_raises(error):
+    """Netmiko reports Paramiko's connection errors as its own timeout, raised while handling the original."""
+    wrapped = NetmikoTimeoutException('A paramiko SSHException occurred during connection creation')
+    wrapped.__context__ = error
+    return wrapped
+
+
+@pytest.mark.parametrize(
+    'error',
+    [
+        BadHostKeyException('192.0.2.1', FakeHostKey(), FakeHostKey()),
+        SSHException("Server '192.0.2.1' not found in known_hosts"),
+    ],
+)
+def test_a_device_presenting_another_host_key_is_refused_for_good(netmiko, error):
+    netmiko({}, error=_as_netmiko_raises(error))
+
+    with pytest.raises(PermanentError, match='did not present its pinned SSH host key') as raised:
+        collect([], host_key=HOST_KEY)
+
+    assert raised.value.reason == 'auth'
+
+
+def test_without_a_pinned_key_a_host_key_error_is_just_a_failed_connection(netmiko):
+    netmiko({}, error=_as_netmiko_raises(SSHException("Server '192.0.2.1' not found in known_hosts")))
+
+    with pytest.raises(TransientError) as raised:
+        collect([])
+
+    assert raised.value.reason == 'connection'