85e019783b
- Introduced tests for case creation commands, ensuring proper handling of UUIDs and payloads. - Added tests for v5 assignment functionality, validating employee identification and group assignment. - Implemented tests for remote recovery scenarios, including handling network failures and response validation. - Created tests for hardware label generation and validation, ensuring proper URL formatting and QR code generation. - Developed extensive tests for remote control functionality, covering connection isolation, command execution, and user permissions. - Added tests for workflow timer registration, ensuring correct handling of timer states and ownership validation.
93 lines
4.2 KiB
Python
93 lines
4.2 KiB
Python
"""Remote credentials are deliberately separate from normal Hub authentication."""
|
|
import hashlib
|
|
import hmac
|
|
import math
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from fastapi import HTTPException, Request, Response
|
|
from app.core.config import settings
|
|
from app.core.auth_service import AuthService
|
|
from app.core.database import atomic_transaction, execute_query, execute_query_single
|
|
|
|
COOKIE = 'hub_remote_device'
|
|
CSRF_COOKIE = 'hub_remote_csrf'
|
|
|
|
|
|
def today(now=None):
|
|
return (now or datetime.now(timezone.utc)).astimezone(ZoneInfo('Europe/Copenhagen')).date()
|
|
|
|
|
|
def digest(value):
|
|
return hashlib.sha256(value.encode()).hexdigest()
|
|
|
|
|
|
def lookup(value):
|
|
return hmac.new(settings.REMOTE_LOOKUP_KEY.encode(), value.encode(), hashlib.sha256).hexdigest()
|
|
|
|
|
|
def new_code():
|
|
return f'{secrets.randbelow(10**12):012d}'
|
|
|
|
|
|
def guard(request: Request, response: Response):
|
|
response.headers['Cache-Control'] = 'no-store'
|
|
if not settings.REMOTE_ENABLED:
|
|
raise HTTPException(404, 'Fjernbetjening er slået fra')
|
|
if len(settings.REMOTE_LOOKUP_KEY) < 32:
|
|
raise HTTPException(503, 'Fjernbetjening mangler sikker serverkonfiguration')
|
|
if request.method not in {'GET', 'HEAD', 'OPTIONS'}:
|
|
origin = str(request.base_url).rstrip('/')
|
|
if request.headers.get('origin') != origin:
|
|
raise HTTPException(403, 'Ugyldig oprindelse')
|
|
cookie = request.cookies.get(CSRF_COOKIE, '')
|
|
token = request.headers.get('x-remote-csrf', '')
|
|
if not cookie or not hmac.compare_digest(cookie, token):
|
|
raise HTTPException(403, 'Opdatér siden og prøv igen')
|
|
|
|
|
|
def reserve_attempt(ip, user_id=None, device_id=None):
|
|
"""Commit even failed attempts, and serialize all buckets across worker processes."""
|
|
buckets = sorted({lookup('ip:' + ip), *([lookup('user:' + str(user_id))] if user_id else []),
|
|
*([lookup('device:' + str(device_id))] if device_id else [])})
|
|
blocked = None
|
|
with atomic_transaction():
|
|
for bucket in buckets:
|
|
execute_query("INSERT INTO remote_attempts(bucket,next_allowed_at) VALUES(%s,'epoch') ON CONFLICT DO NOTHING", (bucket,))
|
|
rows = execute_query('SELECT next_allowed_at FROM remote_attempts WHERE bucket=ANY(%s) ORDER BY bucket FOR UPDATE', (buckets,))
|
|
now = execute_query_single('SELECT clock_timestamp() AS now')['now']
|
|
wait = max([(row['next_allowed_at'] - now).total_seconds() for row in rows] + [0])
|
|
if wait > 0:
|
|
blocked = max(1, math.ceil(wait))
|
|
else:
|
|
execute_query("UPDATE remote_attempts SET next_allowed_at=clock_timestamp()+INTERVAL '30 seconds' WHERE bucket=ANY(%s)", (buckets,))
|
|
if blocked:
|
|
raise HTTPException(429, 'Vent før næste loginforsøg', headers={'Retry-After': str(blocked)})
|
|
|
|
|
|
def remote_device(request, unlocked=True):
|
|
token = request.cookies.get(COOKIE)
|
|
if not token:
|
|
raise HTTPException(401, 'Forbind enheden med din kode')
|
|
row = execute_query_single('''SELECT d.*,r.user_id,p.generation,p.pin_hash,u.username,u.full_name,u.is_superadmin
|
|
FROM remote_devices d JOIN remote_controls r ON r.id=d.remote_id
|
|
JOIN remote_pins p ON p.user_id=r.user_id JOIN users u ON u.user_id=r.user_id
|
|
WHERE d.token_hash=%s AND NOT d.revoked AND NOT r.revoked AND d.expires_at>NOW() AND u.is_active=TRUE''', (digest(token),))
|
|
if not row:
|
|
raise HTTPException(401, 'Forbindelsen er udløbet eller tilbagekaldt')
|
|
if unlocked and (row['unlocked_date'] != today() or row['pin_generation'] != row['generation']):
|
|
raise HTTPException(423, 'Indtast dagens PIN')
|
|
return row
|
|
|
|
|
|
def actor(device):
|
|
# Current permissions are fetched for every request; no cached admin JWT is minted.
|
|
return {'id': device['user_id'], 'username': device['username'], 'full_name': device['full_name'],
|
|
'is_superadmin': device['is_superadmin'], 'permissions': AuthService.get_user_permissions(device['user_id'])}
|
|
|
|
|
|
def case_permission(user):
|
|
if not user.get('is_superadmin') and not {'cases.view', 'tickets.view'}.intersection(user.get('permissions') or []):
|
|
raise HTTPException(403, 'Du har ikke adgang til sager')
|