feat(security): 本地控制台签名的备份退役证明（#68 切片）

docs/CUSTODIAL_HISTORY.md

@@ -90,11 +90,25 @@ Test Root, rewraps, locally decrypts, ACKs and finalizes it. It refuses a produc
 application without the isolated selftest namespace.
 
 `history_recoverable` covers **enumerated and client-verified revisions only**.
-`online_key_retired` describes online vault material only. `backup_retired` and
-`server_key_retired` remain false until a separate independently verified offline
-retirement procedure exists. Neither a boolean supplied by a network caller nor a
-green test run can establish that old backups lost a key. Restore/revocation replay
-and real operator acceptance are tracked separately in issues #69 and #70.
+`online_key_retired` describes online vault material only. `backup_retired` stays
+false unless a root-signed backup retirement record was imported at the physical
+console (`RootAdmin.sign_backup_retirement` / `RootAdmin.import_backup_retirement`,
+both behind `require_local_console`). The record binds the subject, both old key
+IDs, a non-empty list of named backup sets and a validity window of at most 400
+days; it is signed with purpose `custodial-backup-retirement-v1`. Every read
+re-verifies it against the active, unrevoked root credential, so an altered,
+expired, misbound or pre-rotation record reports `backup_retired=false`
+(`backup_status=local_attestation_invalid|expired`). `server_key_retired` is true
+only when that record and observable online retirement both hold.
+
+The record is an operator attestation for the **listed backup sets only**. It
+does not prove that no other copy exists and it does not claim that production
+backups were destroyed. No network operation writes, accepts or forwards it; a
+`backup_retirement` value inside the network-writable migration body is ignored.
+Neither a boolean supplied by a network caller nor a green test run can establish
+that old backups lost a key. There is no `msg-admin` CLI wiring yet. Restore/
+revocation replay and real operator acceptance are tracked separately in issues
+#69 and #70.
 
 ## Published contract compatibility
 

docs/ISSUES_68_70_PROGRESS.md

@@ -21,8 +21,10 @@ only the challenge's new signing key may inspect its completed result. The
 read reports `identity_switched`, `history_recoverable`, `online_retired`,
 `backup_retired`, `server_key_retired` and a separate `completion_status`.
 Online retirement requires the new primary keys, retired old keys, revoked old
-signing credential and empty destroyed vault fields. No backup-destruction
-attestation exists yet, so `backup_retired` and `server_key_retired` remain false.
+signing credential and empty destroyed vault fields. `backup_retired` becomes true
+only for one exact root-signed record imported at the physical console and
+re-verified on each read (see `CUSTODIAL_HISTORY.md`); `server_key_retired`
+additionally needs online retirement. Without that record both remain false.
 The legacy `status=completed` from an empty-inventory identity switch is not a
 claim that every server-held backup has been destroyed; use the independent
 retirement fields.

src/msg/admin/backup_retirement.py

@@ -0,0 +1,56 @@
+"""Local import of a root-signed custodial backup retirement record.
+
+Never imported by a network adapter, plugin or worker. RootAdmin calls these use
+cases only after require_local_console(); tests call them against disposable
+databases. The record is verified again on every read, so this import is not a
+trusted cache of a success flag.
+"""
+from __future__ import annotations
+
+from msg.constants import ROOT_SUBJECT
+from msg.core.codec import digest, loads
+from msg.core.errors import require
+from msg.core.models import AuditEvent, Event, ResourceRef
+from msg.plugins.common import new_id
+from msg.security.backup_retirement import FIELDS, DOMAIN, check, root_verifier, setting_key
+
+
+def unsigned_statement(value):
+    require(isinstance(value, dict) and set(value) == FIELDS and value.get('domain') == DOMAIN,
+            'backup_retirement_invalid')
+    return value
+
+
+def completed_migration(tx, subject_id, old_identity_key_id):
+    """The one completed migration of this subject that retired exactly this old key."""
+    matches = [loads(body) for (body,) in tx.rows(
+        "SELECT body FROM custodial_upgrades WHERE subject=? AND status='completed' ORDER BY id",
+        (subject_id,)) if loads(body).get('old_identity_key_id') == old_identity_key_id]
+    require(len(matches) == 1, 'backup_retirement_migration_not_found')
+    return matches[0]
+
+
+async def import_record(tx, record, *, now, operator):
+    body = record.get('statement') if isinstance(record, dict) else None
+    require(isinstance(body, dict) and isinstance(body.get('subject_id'), str) and
+            isinstance(body.get('old_identity_key_id'), str), 'backup_retirement_invalid')
+    subject = body['subject_id']
+    details = completed_migration(tx, subject, body['old_identity_key_id'])
+    attestation = check(record, await root_verifier(tx), subject, details, now=now)
+    key = setting_key(subject, attestation.old_identity_key_id)
+    previous = tx.setting(key)
+    tx.set_setting(key, record)
+    data = {'subject_id': subject, 'old_identity_key_id': attestation.old_identity_key_id,
+            'old_encryption_key_id': attestation.old_encryption_key_id,
+            'backup_sets': list(attestation.backup_sets), 'record_digest': attestation.record_digest,
+            'signer_key_id': attestation.signer_key_id, 'operator': operator,
+            'scope': 'listed_backup_sets_only'}
+    event = Event(id=new_id('audit'), type='root.custodial.backup_retirement', time=now,
+        request_id=new_id('local'), actor=ROOT_SUBJECT, subject=subject,
+        resources=(ResourceRef(id=subject),), data=data)
+    await tx.append_audit(AuditEvent(event=event,
+        authority=(ResourceRef(id=tx.setting('active_root_certificate')),),
+        before_digest=digest(previous) if previous is not None else None,
+        after_digest=attestation.record_digest, previous_digest=None, entry_digest='',
+        result='imported'))
+    return attestation

src/msg/admin/root.py

@@ -262,6 +262,43 @@ async def read():
         return asyncio.run(_revoke(app,certificate_id,Ed25519Signer.from_bytes(private),reason=reason,operator=operator))
 
 
+    def sign_backup_retirement(self,source,destination):
+        """Root-sign an operator's statement that listed backup sets lost one old key."""
+        require_local_console(self.config_dir)
+        from msg.admin.backup_retirement import unsigned_statement
+        from msg.security.backup_retirement import sign_record
+        target=Path(destination)
+        require(not target.exists() and not target.is_symlink(),'backup_retirement_destination_exists')
+        body=unsigned_statement(loads(Path(source).read_bytes()))
+        print(canonical({'statement':body,'digest':digest(body),
+                         'claim':'listed_backup_sets_only'}).decode())
+        require(input('Type ATTEST BACKUP RETIREMENT '+digest(body)+': ')==
+                'ATTEST BACKUP RETIREMENT '+digest(body),'approval_cancelled')
+        private=open_private_key(loads(root_envelope(self.config_dir).read_bytes()),getpass.getpass('Root PIN/passphrase: '))
+        record=sign_record(body,Ed25519Signer.from_bytes(private))
+        durable_write(target,canonical(record),mode=0o600)
+        return {'status':'backup_retirement_signed','path':str(target),'digest':digest(record)}
+
+    def import_backup_retirement(self,source):
+        """Only this console path persists a record; reads re-verify it every time."""
+        operator=require_local_console(self.config_dir)
+        from msg.admin.backup_retirement import import_record
+        app=self._app()
+        record=loads(Path(source).read_bytes())
+        print(canonical({'record':record,'digest':digest(record)}).decode())
+        require(input('Type IMPORT BACKUP RETIREMENT '+digest(record)+': ')==
+                'IMPORT BACKUP RETIREMENT '+digest(record),'approval_cancelled')
+        async def run():
+            await app.load()
+            try:
+                async with app.metadata.transaction(write=True) as tx:
+                    return await import_record(tx,record,now=app.clock(),operator=operator)
+            finally:
+                await app.close()
+        attestation=asyncio.run(run())
+        return {'status':'backup_retirement_imported','scope':'listed_backup_sets_only',
+                **attestation.view()}
+
     def backup(self,destination):
         require_local_console(self.config_dir)
         target=Path(destination)

src/msg/plugins/custodial_lifecycle.py

@@ -66,7 +66,7 @@ async def transition(app, ctx, request, tx, subject):
     else:
         require(hmac.compare_digest(details['age_possession_digest'], digest(args['age_proof'])),
                 'age_possession_failed')
-    state = await snapshot(tx, subject.resource_id, row[1], challenge, details)
+    state = await snapshot(tx, subject.resource_id, row[1], challenge, details, now=ctx.now)
     state.require_observation(args)
     before = digest(details)
     if args['action'] == 'refresh':
@@ -142,7 +142,8 @@ async def transition(app, ctx, request, tx, subject):
     await audit(tx, ctx, request, subject.resource_id, 'identity.custodial_' + args['action'],
                 {'challenge_id': args['challenge_id'], 'before_digest': before,
                  'after_digest': digest(details), 'decision_signature': wire(args['migration_ack'])})
-    return HandlerOutput(data=(await snapshot(tx, subject.resource_id, row[1], challenge, details)).data)
+    return HandlerOutput(data=(await snapshot(tx, subject.resource_id, row[1], challenge, details,
+                                              now=ctx.now)).data)
 
 
 async def switch_identity(app, ctx, request, tx, subject, challenge, details, *, retain_decrypt=False):
@@ -204,7 +205,7 @@ async def switch_identity(app, ctx, request, tx, subject, challenge, details, *,
         tx.execute('''UPDATE custodial_vault SET status='destroyed',signing_nonce=NULL,
             signing_ciphertext=NULL,age_nonce=NULL,age_ciphertext=NULL,destroyed_at=? WHERE subject=?''',
             (wire(ctx.now), uid), write=True)
-    current = await snapshot(tx, uid, 'completed', challenge, details)
+    current = await snapshot(tx, uid, 'completed', challenge, details, now=ctx.now)
     completed.update(completion_scope='identity_switch_only', phase=current.data['phase'],
         history_recoverable=current.data['history_recoverable'],
         verification_scope=current.data['verification_scope'],

src/msg/plugins/recovery.py

@@ -40,7 +40,8 @@ async def migration(ctx, request, tx, *, pending=False):
         challenge, details = loads(row[2]), loads(row[3])
         authorize_migration(ctx, subject, row[0], row[1], details)
         require(not pending or row[1] != 'pending', 'custodial_upgrade_not_pending_rewrap')
-        state = await migration_snapshot(tx, subject.resource_id, row[1], challenge, details)
+        state = await migration_snapshot(tx, subject.resource_id, row[1], challenge, details,
+                                         now=ctx.now)
         return subject, row[1], challenge, details, state
 
     @op('identity.custodial_upgrade_inventory',obj({'challenge_id':IDENTIFIER},

src/msg/security/backup_retirement.py

@@ -0,0 +1,135 @@
+"""Root-signed attestation that listed backup sets no longer hold one old key.
+
+Only the physical-console import in msg.admin.backup_retirement persists a
+record, under a settings key that no network operation writes. Every read
+re-verifies the signature against the current, unrevoked root credential and the
+exact subject and old key IDs; an absent, altered, expired or superseded-root
+record is not evidence. The statement is an operator attestation for the listed
+backup sets only, not a proof that no other copy of the key exists anywhere.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
+from msg.constants import ROOT_SUBJECT
+from msg.core.codec import canonical, decode, digest, parse_time, wire
+from msg.core.errors import Failure, require
+from msg.core.models import Signature
+from msg.security.crypto import key_id, verify
+
+DOMAIN = 'msg-custodial-backup-retirement-v1'
+PURPOSE = 'custodial-backup-retirement-v1'
+PREFIX = 'custodial_backup_retirement:'
+MAX_VALIDITY = timedelta(days=400)
+FIELDS = frozenset({'domain', 'subject_id', 'old_identity_key_id', 'old_encryption_key_id',
+                    'backup_sets', 'attested_at', 'not_after', 'operator'})
+
+
+@dataclass(frozen=True)
+class BackupRetirement:
+    """Constructed only by check(); a bare boolean or dict is never accepted."""
+    subject_id: str
+    old_identity_key_id: str
+    old_encryption_key_id: str
+    backup_sets: tuple
+    attested_at: datetime
+    not_after: datetime
+    signer_key_id: str
+    record_digest: str
+
+    def binds(self, subject_id, details):
+        return (self.subject_id == subject_id and
+                self.old_identity_key_id == details.get('old_identity_key_id') and
+                self.old_encryption_key_id == details.get('old_encryption_key_id'))
+
+    def view(self):
+        return {'scope': 'listed_backup_sets_only', 'backup_sets': list(self.backup_sets),
+                'attested_at': wire(self.attested_at), 'not_after': wire(self.not_after),
+                'signer_key_id': self.signer_key_id, 'record_digest': self.record_digest}
+
+
+def setting_key(subject_id, old_identity_key_id):
+    return PREFIX + subject_id + ':' + old_identity_key_id
+
+
+def statement(subject_id, details, backup_sets, *, attested_at, not_after, operator):
+    return {'domain': DOMAIN, 'subject_id': subject_id,
+            'old_identity_key_id': details['old_identity_key_id'],
+            'old_encryption_key_id': details['old_encryption_key_id'],
+            'backup_sets': sorted(set(backup_sets)), 'attested_at': wire(attested_at),
+            'not_after': wire(not_after), 'operator': operator}
+
+
+def sign_record(body, signer):
+    return {'statement': body, 'signature': wire(signer.sign(canonical(body), purpose=PURPOSE))}
+
+
+async def root_verifier(tx):
+    """The key of the active, unrevoked root certificate; rotation retires old records."""
+    try:
+        certificate_id = tx.setting('active_root_certificate')
+        certificate = await tx.certificate(certificate_id)
+        credential = await tx.credential(certificate.key_id)
+        require(certificate.subject_id == ROOT_SUBJECT and
+                not await tx.certificate_revoked(certificate_id) and
+                credential.subject_id == ROOT_SUBJECT and credential.kind == 'signing_key' and
+                credential.revoked_at is None and
+                key_id(credential.verifier) == credential.id == certificate.key_id,
+                'backup_retirement_root_unavailable')
+    except (Failure, KeyError, TypeError, ValueError) as exc:
+        raise Failure('backup_retirement_root_unavailable') from exc
+    return credential.verifier
+
+
+def _sets(value):
+    return (isinstance(value, list) and bool(value) and value == sorted(set(value)) and
+            all(isinstance(item, str) and 0 < len(item) <= 256 for item in value))
+
+
+def check(record, verifier, subject_id, details, *, now):
+    """Return the attestation only if this exact record is valid now for these old keys."""
+    try:
+        require(isinstance(record, dict) and set(record) == {'statement', 'signature'},
+                'backup_retirement_invalid')
+        body = record['statement']
+        require(isinstance(body, dict) and set(body) == FIELDS and body['domain'] == DOMAIN and
+                all(isinstance(body[name], str) and body[name] for name in
+                    ('subject_id', 'old_identity_key_id', 'old_encryption_key_id', 'operator')) and
+                _sets(body['backup_sets']), 'backup_retirement_invalid')
+        attested, not_after = parse_time(body['attested_at']), parse_time(body['not_after'])
+        signature = decode(Signature, record['signature'])
+        verify(verifier, canonical(body), signature, purpose=PURPOSE)
+        require(attested < not_after <= attested + MAX_VALIDITY, 'backup_retirement_invalid')
+    except (Failure, KeyError, TypeError, ValueError) as exc:
+        raise Failure('backup_retirement_invalid') from exc
+    require(isinstance(details, dict) and body['subject_id'] == subject_id and
+            body['old_identity_key_id'] == details.get('old_identity_key_id') and
+            body['old_encryption_key_id'] == details.get('old_encryption_key_id'),
+            'backup_retirement_binding_mismatch')
+    try:
+        current = attested <= now < not_after
+    except TypeError:
+        current = False
+    require(current, 'backup_retirement_expired')
+    return BackupRetirement(subject_id=subject_id,
+        old_identity_key_id=body['old_identity_key_id'],
+        old_encryption_key_id=body['old_encryption_key_id'],
+        backup_sets=tuple(body['backup_sets']), attested_at=attested, not_after=not_after,
+        signer_key_id=signature.key_id, record_digest=digest(record))
+
+
+async def verified(tx, subject_id, details, *, now):
+    """(status, attestation-or-None); any failure is reported as not retired."""
+    old = details.get('old_identity_key_id') if isinstance(details, dict) else None
+    if not isinstance(subject_id, str) or not isinstance(old, str) or not old:
+        return 'unavailable', None
+    record = tx.setting(setting_key(subject_id, old))
+    if record is None:
+        return 'unavailable', None
+    if now is None:
+        return 'clock_unavailable', None
+    try:
+        return 'verified', check(record, await root_verifier(tx), subject_id, details, now=now)
+    except Failure as exc:
+        return ('expired' if exc.code == 'backup_retirement_expired' else 'invalid'), None