File size: 4,099 Bytes
a38ca42 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """
VTX beta key schema, migration, and validation helpers.
"""
from __future__ import annotations
import secrets
import string
from datetime import date
from typing import Any
VALID_STATUSES = frozenset({"ACTIVE", "REVOKED", "EXPIRED", "BANNED", "INVALID"})
KEY_CHARS = string.ascii_uppercase + string.digits
NEW_KEY_PREFIX = "VTX-"
def today_iso() -> str:
return date.today().isoformat()
def _random_key_segment(length: int = 6) -> str:
return "".join(secrets.choice(KEY_CHARS) for _ in range(length))
def generate_unique_key(existing: dict[str, Any]) -> str:
for _ in range(5000):
candidate = f"{NEW_KEY_PREFIX}{_random_key_segment(6)}"
if candidate not in existing:
return candidate
raise RuntimeError("Unable to generate a unique beta key.")
def normalize_record(key: str, record: dict[str, Any] | None) -> dict[str, Any]:
data = dict(record or {})
key_text = str(data.get("key") or key or "").strip()
active = bool(data.get("active"))
status = str(data.get("status") or "").strip().upper()
if not status:
status = "ACTIVE" if active else "REVOKED"
if status not in VALID_STATUSES:
status = "ACTIVE" if active else "REVOKED"
if status != "ACTIVE":
active = False
created = str(data.get("created") or data.get("created_date") or "").strip()
last_activation = data.get("last_activation")
if last_activation is not None and last_activation != "":
last_activation = str(last_activation).strip()
else:
last_activation = None
return {
"key": key_text,
"name": str(data.get("name") or data.get("owner") or "").strip(),
"notes": str(data.get("notes") or "").strip(),
"created": created,
"last_activation": last_activation,
"active": active,
"status": status,
}
def migrate_keys_payload(payload: Any) -> dict[str, dict[str, Any]]:
if payload is None:
return {}
if isinstance(payload, list):
migrated: dict[str, dict[str, Any]] = {}
for item in payload:
if not isinstance(item, dict):
continue
key = str(item.get("key") or "").strip()
if not key:
continue
migrated[key] = normalize_record(key, item)
return migrated
if not isinstance(payload, dict):
return {}
if "keys" in payload and isinstance(payload.get("keys"), list):
return migrate_keys_payload(payload["keys"])
migrated: dict[str, dict[str, Any]] = {}
for raw_key, raw_record in payload.items():
key = str(raw_key or "").strip()
if not key:
continue
if isinstance(raw_record, str):
migrated[key] = normalize_record(
key,
{"key": key, "active": True, "status": "ACTIVE", "notes": raw_record},
)
continue
if not isinstance(raw_record, dict):
continue
migrated[key] = normalize_record(key, raw_record)
return migrated
def validation_result(record: dict[str, Any] | None) -> dict[str, Any]:
if not record:
return {"valid": False, "active": False, "status": "INVALID"}
normalized = normalize_record(str(record.get("key") or ""), record)
active = bool(normalized.get("active"))
status = str(normalized.get("status") or "").strip().upper()
if status == "ACTIVE" and active:
return {"valid": True, "active": True, "status": "ACTIVE"}
if status == "REVOKED" or (not active and status == "ACTIVE"):
return {"valid": True, "active": False, "status": "REVOKED"}
if status in {"EXPIRED", "BANNED"}:
return {"valid": True, "active": False, "status": status}
return {"valid": True, "active": active, "status": status}
def new_record(name: str, notes: str = "") -> dict[str, Any]:
return {
"name": str(name or "").strip(),
"notes": str(notes or "").strip(),
"created": today_iso(),
"last_activation": None,
"active": True,
"status": "ACTIVE",
} |