Codex
156 lines
8.6 KiB
Python
156 lines
8.6 KiB
Python
"""Local accounts. Passwords never appear in public account representations."""
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import re
|
|
import secrets
|
|
import time
|
|
|
|
|
|
class Accounts:
|
|
def __init__(self, app):
|
|
self.a = app
|
|
self.password_attempts = {}
|
|
|
|
def initialize(self):
|
|
a = self.a
|
|
with a.LOCK:
|
|
a.DB.execute('CREATE TABLE IF NOT EXISTS users (id TEXT PRIMARY KEY, body TEXT NOT NULL)')
|
|
if 'user_id' not in [r['name'] for r in a.DB.execute('PRAGMA table_info(sessions)')]:
|
|
a.DB.execute('ALTER TABLE sessions ADD COLUMN user_id TEXT')
|
|
legacy = a.setting('account')
|
|
if legacy and not a.objects('users'):
|
|
user = dict(id=secrets.token_hex(8), username='admin', name='管理员', role='admin',
|
|
disabled=False, siteIds=[], familyAccess='edit', personId='',
|
|
relatedToId='', relationship='', **legacy)
|
|
a.DB.execute('INSERT INTO users VALUES (?,?)', (user['id'], json.dumps(user)))
|
|
a.DB.execute('UPDATE sessions SET user_id=? WHERE user_id IS NULL', (user['id'],))
|
|
a.DB.execute("DELETE FROM settings WHERE key='account'")
|
|
a.DB.commit()
|
|
|
|
@staticmethod
|
|
def public(user):
|
|
return {k: v for k, v in user.items() if k not in ('salt', 'hash')} if user else None
|
|
|
|
def password(self, password):
|
|
if not isinstance(password, str) or not 8 <= len(password) <= 128:
|
|
raise self.a.Problem('密码需要 8 至 128 个字符')
|
|
salt = secrets.token_hex(16)
|
|
return {'salt': salt, 'hash': hashlib.scrypt(password.encode(), salt=salt.encode(), n=16384, r=8, p=1).hex()}
|
|
|
|
def verify(self, username, password):
|
|
if not isinstance(username, str) or not isinstance(password, str) or len(password) > 128:
|
|
return None
|
|
user = next((u for u in self.a.objects('users') if u['username'] == username.strip().lower()), None)
|
|
# Also perform derivation for nonexistent accounts.
|
|
salt = user['salt'] if user else '0' * 32
|
|
derived = hashlib.scrypt(password.encode(), salt=salt.encode(), n=16384, r=8, p=1).hex()
|
|
return user if user and not user['disabled'] and hmac.compare_digest(derived, user['hash']) else None
|
|
|
|
def save(self, data, actor):
|
|
a = self.a
|
|
with a.LOCK:
|
|
old = a.get_object('users', data.get('id'))
|
|
if data.get('id') and not old:
|
|
raise a.Problem('账号不存在', 404)
|
|
uid = old['id'] if old else secrets.token_hex(8)
|
|
username = a.clean_text(data.get('username', ''), 32, True).lower()
|
|
if not re.fullmatch(r'[a-z0-9][a-z0-9_.-]{2,31}', username):
|
|
raise a.Problem('用户名使用 3 至 32 位英文字母、数字、点、下划线或短横线')
|
|
if any(u['username'] == username and u['id'] != uid for u in a.objects('users')):
|
|
raise a.Problem('用户名已存在', 409)
|
|
role = data.get('role', 'member')
|
|
access = data.get('familyAccess', 'none')
|
|
if role not in ('admin', 'member') or access not in ('none', 'read', 'edit'):
|
|
raise a.Problem('权限选项不正确')
|
|
sites = data.get('siteIds', [])
|
|
if not isinstance(sites, list) or any(not isinstance(s, str) or not a.get_object('sites', s) for s in sites):
|
|
raise a.Problem('请选择有效空间')
|
|
ptz_control = data.get('ptzControl', old.get('ptzControl', False) if old else False)
|
|
if not isinstance(ptz_control, bool):
|
|
raise a.Problem('云台权限格式不正确')
|
|
disabled = data.get('disabled', False)
|
|
if not isinstance(disabled, bool):
|
|
raise a.Problem('账号状态不正确')
|
|
if old and old['role'] == 'admin' and (role != 'admin' or disabled):
|
|
if not any(u['id'] != uid and u['role'] == 'admin' and not u['disabled'] for u in a.objects('users')):
|
|
raise a.Problem('至少保留一个启用的管理员')
|
|
if actor and actor['id'] == uid and (disabled or role != actor['role']):
|
|
raise a.Problem('不能停用或降低当前登录账号的管理权限')
|
|
related = a.clean_text(data.get('relatedToId', ''), 32)
|
|
person = a.clean_text(data.get('personId', ''), 32)
|
|
if related and related != uid and not a.get_object('users', related):
|
|
raise a.Problem('关系参照账号不存在')
|
|
if person:
|
|
if not a.get_object('people', person):
|
|
raise a.Problem('家谱人物不存在')
|
|
if any(u.get('personId') == person and u['id'] != uid for u in a.objects('users')):
|
|
raise a.Problem('这个人物已经关联其他账号')
|
|
password = data.get('password', '')
|
|
if not isinstance(password, str):
|
|
raise a.Problem('密码格式不正确')
|
|
credentials = self.password(password) if password or not old else {k: old[k] for k in ('salt', 'hash')}
|
|
user = dict(id=uid, username=username, name=a.clean_text(data.get('name', username), 80, True),
|
|
role=role, disabled=disabled, siteIds=sorted(set(sites)), familyAccess=access, ptzControl=ptz_control,
|
|
personId=person, relatedToId=related,
|
|
relationship=a.clean_text(data.get('relationship', ''), 40), **credentials)
|
|
# Revoke sessions when credentials or authorization change, not for label edits.
|
|
if old and any(old.get(k) != user.get(k) for k in ('hash', 'role', 'siteIds', 'familyAccess', 'disabled', 'username', 'ptzControl')):
|
|
a.DB.execute('DELETE FROM sessions WHERE user_id=?', (uid,))
|
|
a.save_object('users', user)
|
|
a.audit('更新账号' if old else '建立账号', username)
|
|
return self.public(user)
|
|
|
|
def change_password(self, data, actor):
|
|
a = self.a
|
|
with a.LOCK:
|
|
current = a.get_object('users', actor['id'])
|
|
if not current or current['disabled']:
|
|
raise a.Problem('请重新登录', 401)
|
|
now = time.time()
|
|
self.password_attempts = {k: v for k, v in self.password_attempts.items() if v[1] > now}
|
|
count, expiry = self.password_attempts.get(current['id'], (0, now + 900))
|
|
if count >= 8:
|
|
raise a.Problem('当前密码尝试次数过多，请 15 分钟后重试', 429)
|
|
if not self.verify(current['username'], data.get('currentPassword')):
|
|
self.password_attempts[current['id']] = (count + 1, expiry)
|
|
raise a.Problem('当前密码不正确', 403)
|
|
if data.get('newPassword') == data.get('currentPassword'):
|
|
raise a.Problem('新密码不能与当前密码相同')
|
|
credentials = self.password(data.get('newPassword'))
|
|
with a.DB:
|
|
a.DB.execute('UPDATE users SET body=? WHERE id=?', (json.dumps(dict(current, **credentials)), current['id']))
|
|
a.DB.execute('DELETE FROM sessions WHERE user_id=?', (current['id'],))
|
|
self.password_attempts.pop(current['id'], None)
|
|
a.audit('本人修改密码', current['username'])
|
|
return {'ok': True}
|
|
|
|
def camera(self, user, camera):
|
|
if not camera:
|
|
raise self.a.Problem('摄像头不存在', 404)
|
|
if user['role'] != 'admin' and camera.get('siteId') not in user['siteIds']:
|
|
raise self.a.Problem('没有这个空间的访问权限', 403)
|
|
return camera
|
|
|
|
def control(self, user, camera):
|
|
self.camera(user, camera)
|
|
if user['role'] != 'admin' and not user.get('ptzControl', False):
|
|
raise self.a.Problem('没有云台控制权限，请联系管理员授权', 403)
|
|
if not camera.get('enabled'):
|
|
raise self.a.Problem('摄像头已停用', 409)
|
|
return camera
|
|
|
|
def state(self, user, state):
|
|
state['user'] = self.public(user)
|
|
if user['role'] == 'admin':
|
|
return state
|
|
permitted = set(user['siteIds'])
|
|
state['sites'] = [s for s in state['sites'] if s['id'] in permitted]
|
|
state['assets'] = [s for s in state['assets'] if s['siteId'] in permitted]
|
|
fields = ('id', 'name', 'siteId', 'assetId', 'point', 'enabled', 'ready', 'recordingActive', 'archiveSource')
|
|
state['cameras'] = [{k: c.get(k) for k in fields} for c in state['cameras'] if c['siteId'] in permitted]
|
|
state['recorders'] = []
|
|
state['scan'] = {}
|
|
state['storage'] = {}
|
|
return state
|