79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
## Copyright © 2026 Olaf Kolkman
|
|
## SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
import time
|
|
from urllib.parse import quote
|
|
from uuid import uuid4
|
|
|
|
from backend.app.database import get_connection
|
|
|
|
|
|
def create_secret() -> str:
|
|
return base64.b32encode(secrets.token_bytes(20)).decode('ascii').rstrip('=')
|
|
|
|
|
|
def create_recovery_codes(user_id: str, count: int = 10) -> list[str]:
|
|
codes = [secrets.token_urlsafe(9) for _ in range(count)]
|
|
with get_connection() as conn:
|
|
conn.execute('DELETE FROM otp_recovery_codes WHERE user_id = ?', (user_id,))
|
|
conn.executemany(
|
|
'INSERT INTO otp_recovery_codes (id, user_id, code_hash) VALUES (?, ?, ?)',
|
|
[(str(uuid4()), user_id, hash_recovery_code(code)) for code in codes],
|
|
)
|
|
conn.commit()
|
|
return codes
|
|
|
|
|
|
def hash_recovery_code(code: str) -> str:
|
|
return hashlib.sha256(code.strip().encode('utf-8')).hexdigest()
|
|
|
|
|
|
def consume_recovery_code(user_id: str, code: str) -> bool:
|
|
with get_connection() as conn:
|
|
cursor = conn.execute(
|
|
'''UPDATE otp_recovery_codes
|
|
SET used = 1, used_at = CURRENT_TIMESTAMP
|
|
WHERE user_id = ? AND code_hash = ? AND used = 0''',
|
|
(user_id, hash_recovery_code(code)),
|
|
)
|
|
conn.commit()
|
|
return cursor.rowcount == 1
|
|
|
|
|
|
def provisioning_uri(secret: str, username: str, issuer: str = 'LinkLog') -> str:
|
|
return f'otpauth://totp/{quote(issuer)}:{quote(username)}?secret={secret}&issuer={quote(issuer)}'
|
|
|
|
|
|
def current_code(secret: str, timestamp: float | None = None) -> str:
|
|
padded_secret = secret + '=' * (-len(secret) % 8)
|
|
key = base64.b32decode(padded_secret, casefold=True)
|
|
counter = int(timestamp if timestamp is not None else time.time()) // 30
|
|
digest = hmac.new(key, counter.to_bytes(8, 'big'), hashlib.sha1).digest()
|
|
index = digest[-1] & 0x0f
|
|
value = (int.from_bytes(digest[index:index + 4], 'big') & 0x7fffffff) % 1_000_000
|
|
return f'{value:06d}'
|
|
|
|
|
|
def verify_code(secret: str | None, code: str | None) -> bool:
|
|
if not secret or not code:
|
|
return False
|
|
normalized_code = code.strip()
|
|
if len(normalized_code) != 6 or not normalized_code.isdigit():
|
|
return False
|
|
padded_secret = secret + '=' * (-len(secret) % 8)
|
|
try:
|
|
key = base64.b32decode(padded_secret, casefold=True)
|
|
except (ValueError, base64.binascii.Error):
|
|
return False
|
|
counter = int(time.time()) // 30
|
|
for offset in (-1, 0, 1):
|
|
digest = hmac.new(key, (counter + offset).to_bytes(8, 'big'), hashlib.sha1).digest()
|
|
index = digest[-1] & 0x0f
|
|
value = (int.from_bytes(digest[index:index + 4], 'big') & 0x7fffffff) % 1_000_000
|
|
if hmac.compare_digest(f'{value:06d}', normalized_code):
|
|
return True
|
|
return False |