Praxis CI
813bd586d6
v0.3 milestone merged to main. Mastery scoring + competency rubrics + verifiable credentials (formative-tier) shipped. 13/13 REQ-IDs covered. Next milestone: v0.4 (operator tier — cohort dashboard + auth + Postgres). ---ci--- project: praxis phase: 2 milestone: v0.3 status: complete milestone_complete: true milestone_merged_to_main: true ---/ci---
75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
"""Bitstring Status List revocation (SLICE-09 TASK-09-03, REQ-NFR-VC-02).
|
|
|
|
W3C Bitstring Status List v1.0 — one bit per issued credential. bit=1 means
|
|
revoked. Persisted in SQLite `status_lists` table. Revocation latency = next
|
|
verify call (no cache — status list fetched from SQLite on every verification,
|
|
per REQ-NFR-VC-02). Minimum 131072-bit (16KB) list for herd privacy per spec.
|
|
|
|
Slot allocation is tracked separately from the revocation bitstring (the
|
|
revocation bit is 0 for a newly-issued active credential, so it cannot
|
|
distinguish "allocated-active" from "never-allocated"). A parallel allocation
|
|
bitstring (`{list_id}_alloc`) records which slots have been handed out.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from db.store import PraxisStore
|
|
|
|
_MIN_BITS = 131072
|
|
|
|
|
|
class BitstringStatusList:
|
|
def __init__(self, store: PraxisStore, list_id: str = "default") -> None:
|
|
self.store = store
|
|
self.list_id = list_id
|
|
self._alloc_id = f"{list_id}_alloc"
|
|
|
|
async def _load(self, list_id: str) -> bytearray:
|
|
row = await self.store.get_status_list(list_id)
|
|
if row is None:
|
|
buf = bytearray(_MIN_BITS // 8)
|
|
await self.store.upsert_status_list(list_id, bytes(buf), _MIN_BITS)
|
|
return buf
|
|
return bytearray(row["bitstring"])
|
|
|
|
async def set_status(self, credential_idx: int, revoked: bool) -> None:
|
|
buf = await self._load(self.list_id)
|
|
byte_pos = credential_idx >> 3
|
|
bit_pos = credential_idx & 7
|
|
if revoked:
|
|
buf[byte_pos] |= 1 << bit_pos
|
|
else:
|
|
buf[byte_pos] &= ~(1 << bit_pos)
|
|
size = len(buf) * 8
|
|
await self.store.upsert_status_list(self.list_id, bytes(buf), size)
|
|
|
|
async def get_status(self, credential_idx: int) -> bool:
|
|
buf = await self._load(self.list_id)
|
|
byte_pos = credential_idx >> 3
|
|
bit_pos = credential_idx & 7
|
|
if byte_pos >= len(buf):
|
|
return False
|
|
return bool((buf[byte_pos] >> bit_pos) & 1)
|
|
|
|
async def allocate_slot(self) -> int:
|
|
buf = await self._load(self._alloc_id)
|
|
for i in range(len(buf) * 8):
|
|
byte_pos = i >> 3
|
|
bit_pos = i & 7
|
|
if not (buf[byte_pos] >> bit_pos) & 1:
|
|
buf[byte_pos] |= 1 << bit_pos
|
|
size = len(buf) * 8
|
|
await self.store.upsert_status_list(
|
|
self._alloc_id, bytes(buf), size
|
|
)
|
|
return i
|
|
new_size = (len(buf) * 8) * 2
|
|
new_buf = bytearray(new_size // 8)
|
|
new_buf[: len(buf)] = buf
|
|
idx = len(buf) * 8
|
|
new_buf[idx >> 3] |= 1 << (idx & 7)
|
|
await self.store.upsert_status_list(self._alloc_id, bytes(new_buf), new_size)
|
|
return idx
|
|
|
|
|
|
__all__ = ["BitstringStatusList"] |