Verify the integrity of an AuditChain. Checks: 1. Each entry's entry_hash is consistent 2. Each entry's prev_hash links to the prior entry 3. Sequence numbers are contiguous 4. head_hash matches the last entry's entry_hash
(
chain_id: str,
db: Annotated[AsyncSession, Depends(get_db)],
)
| 797 | description="Verify the integrity of an AuditChain from the database.", |
| 798 | ) |
| 799 | async def verify_chain( |
| 800 | chain_id: str, |
| 801 | db: Annotated[AsyncSession, Depends(get_db)], |
| 802 | ) -> AuditChainVerifyResponse: |
| 803 | """ |
| 804 | Verify the integrity of an AuditChain. |
| 805 | |
| 806 | Checks: |
| 807 | 1. Each entry's entry_hash is consistent |
| 808 | 2. Each entry's prev_hash links to the prior entry |
| 809 | 3. Sequence numbers are contiguous |
| 810 | 4. head_hash matches the last entry's entry_hash |
| 811 | """ |
| 812 | svc = AuditChainService() |
| 813 | valid = await svc.verify_integrity(db, chain_id) |
| 814 | return AuditChainVerifyResponse( |
| 815 | chain_id=chain_id, |
| 816 | valid=valid, |
| 817 | message="Integrity check passed" if valid else "Integrity check FAILED — tampering detected", |
| 818 | ) |
| 819 | |
| 820 | |
| 821 | @_audit.get( |
nothing calls this directly
no test coverage detected