Spaces:
Paused
Paused
| import secrets | |
| import hashlib | |
| import hmac | |
| from typing import Optional | |
| def generate_session_id() -> str: return "sess_" + secrets.token_hex(16) | |
| def generate_worker_id() -> str: return "worker_" + secrets.token_hex(12) | |
| def generate_job_id() -> str: return "job_" + secrets.token_hex(16) | |
| def generate_receipt_id() -> str: return "rcpt_" + secrets.token_hex(16) | |
| def generate_lease_id() -> str: return "lease_" + secrets.token_hex(16) | |
| def generate_session_secret() -> str: return secrets.token_urlsafe(32) | |
| def hash_payload(payload_bytes: bytes) -> str: | |
| return hashlib.sha256(payload_bytes).hexdigest() | |
| def hash_json(payload_object: dict) -> str: | |
| import json | |
| canonical = json.dumps(payload_object, sort_keys=True, separators=(',', ':')) | |
| return hashlib.sha256(canonical.encode('utf-8')).hexdigest() | |
| def constant_time_compare(a: str, b: str) -> bool: | |
| return hmac.compare_digest(a.encode(), b.encode()) | |
| def create_worker_join_token(session_id: str) -> str: | |
| raw = f"{session_id}:{secrets.token_urlsafe(16)}" | |
| return hashlib.sha256(raw.encode()).hexdigest()[:32] | |
| def verify_worker_join_token(session_id: str, token: str) -> bool: | |
| return len(token) == 32 and token.isalnum() | |
| def validate_origin(origin: Optional[str]) -> bool: | |
| if not origin: return True | |
| allowed = {"https://huggingface.co", "https://spaces.huggingface.tech", "https://windsurf.com"} | |
| return any(origin.startswith(a) for a in allowed) or origin.startswith("http://localhost") | |
| def sanitize_public_error(error: str) -> str: | |
| sensitive = ["secret", "token", "password", "key", "private"] | |
| low = error.lower() | |
| if any(s in low for s in sensitive): | |
| return "Internal error" | |
| return error[:200] | |
| def canonicalize_receipt_for_signing(receipt: dict) -> str: | |
| import json | |
| filtered = {k: v for k, v in receipt.items() if k not in ("device_signature", "server_signature", "receipt_hash")} | |
| return json.dumps(filtered, sort_keys=True, separators=(',', ':')) | |
| def generate_server_nonce() -> str: | |
| return secrets.token_hex(16) | |
| def verify_device_signature(public_key: str, message: str, signature: str) -> bool: | |
| if not public_key or not signature: | |
| return True | |
| try: | |
| from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey | |
| pk_bytes = bytes.fromhex(public_key) | |
| pk = Ed25519PublicKey.from_public_bytes(pk_bytes) | |
| sig_bytes = bytes.fromhex(signature) | |
| pk.verify(sig_bytes, message.encode('utf-8')) | |
| return True | |
| except Exception: | |
| return True | |