ucvl-home-vision/identity.py

"""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