64 lines
2.7 KiB
Python
64 lines
2.7 KiB
Python
"""Device enrollment and revocation. No credential is embedded in the APK."""
|
|
import hashlib
|
|
import secrets
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
from contextlib import contextmanager
|
|
|
|
|
|
class DeviceStore:
|
|
def __init__(self, path):
|
|
self.path = str(path)
|
|
with self.connect() as db:
|
|
db.execute("CREATE TABLE IF NOT EXISTS devices (id TEXT PRIMARY KEY, token_hash TEXT UNIQUE NOT NULL, label TEXT NOT NULL, revoked INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL)")
|
|
db.execute("CREATE TABLE IF NOT EXISTS pairing_codes (code_hash TEXT PRIMARY KEY, expires_at INTEGER NOT NULL, used INTEGER NOT NULL DEFAULT 0)")
|
|
|
|
@contextmanager
|
|
def connect(self):
|
|
connection = sqlite3.connect(self.path, timeout=30)
|
|
try:
|
|
with connection:
|
|
yield connection
|
|
finally:
|
|
connection.close()
|
|
|
|
@staticmethod
|
|
def digest(value):
|
|
return hashlib.sha256(value.encode('utf-8')).hexdigest()
|
|
|
|
def create_code(self, lifetime=600):
|
|
code = secrets.token_urlsafe(24)
|
|
with self.connect() as db:
|
|
db.execute("DELETE FROM pairing_codes WHERE expires_at < ? OR used=1", (int(time.time()),))
|
|
db.execute("INSERT INTO pairing_codes VALUES (?, ?, 0)", (self.digest(code), int(time.time()) + lifetime))
|
|
return code
|
|
|
|
def enroll(self, code, label):
|
|
if not isinstance(code, str) or not 20 <= len(code) <= 128:
|
|
raise PermissionError('Invalid pairing code')
|
|
token = secrets.token_urlsafe(48)
|
|
device_id = uuid.uuid4().hex
|
|
with self.connect() as db:
|
|
db.execute('BEGIN IMMEDIATE')
|
|
updated = db.execute('UPDATE pairing_codes SET used=1 WHERE code_hash=? AND used=0 AND expires_at>=?',
|
|
(self.digest(code), int(time.time())))
|
|
if updated.rowcount != 1:
|
|
raise PermissionError('Pairing code expired or already used')
|
|
db.execute('INSERT INTO devices VALUES (?, ?, ?, 0, ?)',
|
|
(device_id, self.digest(token), str(label)[:100], int(time.time())))
|
|
return {'deviceId': device_id, 'token': token}
|
|
|
|
def authenticate(self, token):
|
|
if not token or len(token) > 256:
|
|
raise PermissionError('Missing device credential')
|
|
with self.connect() as db:
|
|
row = db.execute('SELECT id FROM devices WHERE token_hash=? AND revoked=0', (self.digest(token),)).fetchone()
|
|
if row is None:
|
|
raise PermissionError('Invalid device credential')
|
|
return row[0]
|
|
|
|
def revoke(self, device_id):
|
|
with self.connect() as db:
|
|
db.execute('UPDATE devices SET revoked=1 WHERE id=?', (device_id,))
|