nngo
b84a6d9245
CI / Python lint (flake8) (push) Has been cancelled
CI / Python syntax check (push) Has been cancelled
CI / Alembic migration chain (push) Has been cancelled
CI / JavaScript syntax check (push) Has been cancelled
CI / Pytest (push) Has been cancelled
CI / Build extension zip (push) Has been cancelled
104 lines
4.6 KiB
Python
104 lines
4.6 KiB
Python
from datetime import datetime, timezone, timedelta
|
|
|
|
from sqlalchemy.dialects.mysql import INTEGER
|
|
|
|
from app import db
|
|
|
|
|
|
class EmergencyAccess(db.Model):
|
|
"""
|
|
Emergency access grant from a vault owner (grantor) to a trusted contact (grantee).
|
|
|
|
Status flow:
|
|
invited → grantee calls /accept → accepted
|
|
accepted → grantor calls /provide → ready (enc_vault stored)
|
|
ready → grantee calls /request → pending (wait timer starts)
|
|
pending → grantor calls /deny → ready (reset, grantee can request again)
|
|
pending (wait_days elapsed) → grantable (grantee fetches vault)
|
|
|
|
Retrieval does not change `status`: the grant stays 'pending' so the grantor
|
|
keeps seeing it as active and can revoke it. What retrieval does change is
|
|
vault_retrieved_at / vault_retrieval_count, which the grantor's UI surfaces.
|
|
|
|
Zero-knowledge: enc_vault is a JSON array of vault items re-encrypted by the grantor
|
|
using the ECDH shared secret (grantor private key + grantee public key).
|
|
"""
|
|
__tablename__ = 'emergency_access'
|
|
|
|
id = db.Column(INTEGER(unsigned=True), autoincrement=True, primary_key=True)
|
|
grantor_id = db.Column(
|
|
INTEGER(unsigned=True),
|
|
db.ForeignKey('users.id', ondelete='CASCADE'),
|
|
nullable=False,
|
|
)
|
|
grantee_email = db.Column(db.String(255), nullable=False)
|
|
grantee_id = db.Column(
|
|
INTEGER(unsigned=True),
|
|
db.ForeignKey('users.id', ondelete='SET NULL'),
|
|
nullable=True,
|
|
)
|
|
wait_days = db.Column(db.Integer, default=7, nullable=False)
|
|
# invited | accepted | ready | pending | denied
|
|
status = db.Column(db.String(20), default='invited', nullable=False)
|
|
request_initiated_at = db.Column(db.DateTime, nullable=True)
|
|
# JSON string: [{ id, name, item_type, enc_data, iv }, ...]
|
|
enc_vault = db.Column(db.Text, nullable=True)
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc).replace(tzinfo=None), nullable=False)
|
|
# Retrieval tracking — makes grantee access to the snapshot visible to the
|
|
# grantor. Retrieval is not blocked after the first time (the grantor may be
|
|
# unable to re-provision, which is the entire premise of emergency access);
|
|
# the wait period is the gate, and these make use of it auditable.
|
|
vault_retrieved_at = db.Column(db.DateTime, nullable=True)
|
|
vault_retrieval_count = db.Column(db.Integer, default=0, nullable=False, server_default='0')
|
|
|
|
@property
|
|
def wait_elapsed(self):
|
|
"""True if the wait period has passed since the access request."""
|
|
if self.status != 'pending' or not self.request_initiated_at:
|
|
return False
|
|
return datetime.now(timezone.utc).replace(tzinfo=None) >= self.request_initiated_at + timedelta(days=self.wait_days)
|
|
|
|
def to_dict(self, grantor_email=None):
|
|
return {
|
|
'id': self.id,
|
|
'grantor_id': self.grantor_id,
|
|
'grantor_email': grantor_email,
|
|
'grantee_email': self.grantee_email,
|
|
'grantee_id': self.grantee_id,
|
|
'wait_days': self.wait_days,
|
|
'status': self.status,
|
|
'wait_elapsed': self.wait_elapsed,
|
|
'request_initiated_at': (
|
|
self.request_initiated_at.isoformat() if self.request_initiated_at else None
|
|
),
|
|
'created_at': self.created_at.isoformat() if self.created_at else None,
|
|
'vault_retrieved_at': (
|
|
self.vault_retrieved_at.isoformat() if self.vault_retrieved_at else None
|
|
),
|
|
'vault_retrieval_count': self.vault_retrieval_count or 0,
|
|
# True when enc_vault contains items in the old format (has a plaintext
|
|
# 'name' field instead of enc_name/iv_name). Grantor should re-provision.
|
|
'enc_vault_is_legacy': self._enc_vault_is_legacy(),
|
|
}
|
|
|
|
def _enc_vault_is_legacy(self) -> bool:
|
|
"""
|
|
Return True if the stored enc_vault snapshot was created before the
|
|
enc_name migration — i.e. any item has a 'name' key (plaintext) but
|
|
lacks 'enc_name'. Returns False if no snapshot exists or all items
|
|
use the new format.
|
|
"""
|
|
if not self.enc_vault:
|
|
return False
|
|
try:
|
|
import json
|
|
items = json.loads(self.enc_vault)
|
|
if not isinstance(items, list):
|
|
return False
|
|
# Old format: item has 'name' and no 'enc_name'
|
|
return any(
|
|
isinstance(item, dict) and 'name' in item and 'enc_name' not in item
|
|
for item in items
|
|
)
|
|
except (ValueError, TypeError):
|
|
return False |