Codex
212 lines
14 KiB
Python
212 lines
14 KiB
Python
"""Two-household HTTP isolation and management succession regressions."""
|
|
import hashlib
|
|
import http.client
|
|
import json
|
|
import secrets
|
|
import sqlite3
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
# Reuse only module initialization; never inherit the original WebTests.
|
|
from test_app import app
|
|
|
|
|
|
class HouseholdTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
db = sqlite3.connect(':memory:', check_same_thread=False)
|
|
db.row_factory = sqlite3.Row
|
|
self.addCleanup(db.close)
|
|
db.executescript('''
|
|
CREATE TABLE settings (key TEXT PRIMARY KEY,value TEXT NOT NULL);
|
|
CREATE TABLE sessions (hash TEXT PRIMARY KEY,expires REAL NOT NULL);
|
|
CREATE TABLE audit (id INTEGER PRIMARY KEY,at TEXT,action TEXT,name TEXT);
|
|
''')
|
|
for table in ('sites', 'assets', 'recorders', 'cameras'):
|
|
db.execute('CREATE TABLE '+table+' (id TEXT PRIMARY KEY,body TEXT NOT NULL)')
|
|
for attr, value in [('DB', db), ('DATA', Path(self.temp.name)),
|
|
('write_media_config', lambda: None)]:
|
|
p = patch.object(app, attr, value); p.start(); self.addCleanup(p.stop)
|
|
app.initialize()
|
|
self.root = app.ACCOUNTS.save(dict(username='admin', password='fixture-only', role='admin'), None)
|
|
self.aid = self.root['familyId']
|
|
self.bid = app.HOUSEHOLDS.save({'name': '乙家庭'}, self.root)['id']
|
|
self.roots = {h: app.HOUSEHOLDS.context(self.root, h) for h in (self.aid, self.bid)}
|
|
self.managers, self.members, self.sites, self.cameras, self.assets = {}, {}, {}, {}, {}
|
|
for i, home in enumerate((self.aid, self.bid)):
|
|
actor = self.roots[home]
|
|
self.sites[home] = app.save_site({'name': 'home-'+str(i)}, actor)
|
|
self.assets[home] = app.save_asset(dict(name='asset-'+str(i), siteId=self.sites[home]['id']), actor)
|
|
self.cameras[home] = app.save_camera(dict(name='camera-'+str(i), assetId=self.assets[home]['id'],
|
|
host='192.168.3.'+str(10+i), enabled=True), actor)
|
|
self.managers[home] = app.ACCOUNTS.save(dict(username='manager'+str(i), password='fixture-only', role='family_admin'), actor)
|
|
self.members[home] = app.ACCOUNTS.save(dict(username='member'+str(i), password='fixture-only',
|
|
siteIds=[self.sites[home]['id']], familyAccess='edit', ptzControl=True), actor)
|
|
self.owner, self.other = self.managers[self.aid], self.managers[self.bid]
|
|
self.backup = app.ACCOUNTS.save(dict(username='backup', password='fixture-only', role='family_admin'), self.root)
|
|
self.tokens = {}
|
|
for user in [self.root, self.backup, *self.managers.values(), *self.members.values()]:
|
|
token = secrets.token_hex(32); self.tokens[user['id']] = token
|
|
db.execute('INSERT INTO sessions VALUES (?,?,?)', (hashlib.sha256(token.encode()).hexdigest(), time.time()+600, user['id']))
|
|
db.commit()
|
|
def state():
|
|
return {**{t: app.objects(t) for t in ('sites','assets','recorders','cameras')}, 'scan': {'network':'private'}, 'storage': {'maxGB':40}}
|
|
p = patch.object(app, 'status', state); p.start(); self.addCleanup(p.stop)
|
|
self.server = app.ThreadingHTTPServer(('127.0.0.1', 0), app.Handler)
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True); self.thread.start()
|
|
self.addCleanup(self.stop_server)
|
|
|
|
def stop_server(self):
|
|
self.server.shutdown(); self.server.server_close(); self.thread.join()
|
|
|
|
def request(self, path, user=None, data=None, home=None):
|
|
user = user or self.owner
|
|
c = http.client.HTTPConnection(*self.server.server_address, timeout=3)
|
|
headers = {'Cookie': 'vision='+self.tokens[user['id']]}
|
|
if home: headers['X-Household-Id'] = home
|
|
if data is not None: headers['Content-Type'] = 'application/json'
|
|
c.request('GET' if data is None else 'POST', path, json.dumps(data) if data is not None else None, headers)
|
|
response = c.getresponse(); raw = response.read(); code = response.status; c.close()
|
|
return code, json.loads(raw)
|
|
|
|
def leadership(self, user=None, **fields):
|
|
data = dict(id=self.aid, ownerId=self.owner['id'], successorId=self.backup['id'], revision=0, reason='提前安排备用负责人')
|
|
return self.request('/api/households/leadership', user or self.root, data=dict(data, **fields))
|
|
|
|
def test_state_and_accounts_are_scoped(self):
|
|
for home in (self.aid, self.bid):
|
|
for user in (self.managers[home], self.members[home]):
|
|
code, body = self.request('/api/state', user)
|
|
self.assertEqual(code, 200)
|
|
for table in ('sites','assets','cameras'):
|
|
self.assertEqual(len(body[table]), 1)
|
|
self.assertEqual(body['cameras'][0]['id'], self.cameras[home]['id'])
|
|
self.assertEqual([h['id'] for h in body['households']], [home])
|
|
self.assertEqual(body['storage'], {}); self.assertEqual(body['scan'], {})
|
|
code, body = self.request('/api/users')
|
|
self.assertEqual(code, 200)
|
|
self.assertTrue(all(u['familyId']==self.aid and u['role']!='admin' for u in body['users']))
|
|
self.assertNotIn('hash', json.dumps(body)); self.assertNotIn('salt', json.dumps(body))
|
|
|
|
def test_forged_household_header_denied_root_can_switch(self):
|
|
for user in (self.owner, self.members[self.aid]):
|
|
for path in ('/api/session','/api/state','/api/family','/api/households'):
|
|
self.assertEqual(self.request(path, user, home=self.bid)[0], 403)
|
|
code, body = self.request('/api/state', self.root, home=self.bid)
|
|
self.assertEqual(code, 200); self.assertEqual(body['familyId'], self.bid)
|
|
self.assertEqual(body['cameras'][0]['id'], self.cameras[self.bid]['id'])
|
|
|
|
def test_guessed_camera_ids_cannot_access_media_or_controls(self):
|
|
camera = self.cameras[self.bid]['id']
|
|
paths = [f'/media/live/cam_{camera}_{quality}/{file}' for quality in ('main','sub','hd','compat')
|
|
for file in ('index.m3u8','init.mp4','segment1.mp4')]
|
|
paths += [f'/api/recordings?camera={camera}', f'/media/playback?camera={camera}',
|
|
f'/media/playback?camera={camera}&download=1', f'/api/ptz?camera={camera}',
|
|
f'/api/ptz/presets?camera={camera}', f'/api/camera/settings?camera={camera}']
|
|
with patch.object(app.Handler, 'proxy') as proxy:
|
|
for user in (self.owner,self.members[self.aid]):
|
|
for path in paths:
|
|
with self.subTest(path=path): self.assertIn(self.request(path,user)[0], (403,404))
|
|
for path in ('/api/ptz','/api/ptz/presets','/api/camera/settings'):
|
|
self.assertIn(self.request(path,user,dict(camera=camera, action='stop'))[0], (403,404))
|
|
proxy.assert_not_called()
|
|
|
|
def test_own_media_access_and_ungranted_site(self):
|
|
def proxy(handler,*args,**kwargs): handler.answer({'ok':True})
|
|
camera = self.cameras[self.aid]['id']; path = f'/media/live/cam_{camera}_hd/index.m3u8'
|
|
with patch.object(app.Handler,'proxy',proxy):
|
|
self.assertEqual(self.request(path,self.members[self.aid])[0],200)
|
|
member=dict(self.members[self.aid],siteIds=[])
|
|
# Update DB directly to retain token and exercise per-request permission lookup.
|
|
app.save_object('users',dict(app.get_object('users',member['id']),siteIds=[]))
|
|
self.assertEqual(self.request(path,member)[0],403)
|
|
self.assertEqual(self.request(path,self.owner)[0],200)
|
|
|
|
def test_cross_household_edits_references_and_ownership_blocked(self):
|
|
cases=[('/api/sites',dict(self.sites[self.bid],name='stolen')),
|
|
('/api/assets',dict(name='bad',siteId=self.sites[self.bid]['id'])),
|
|
('/api/cameras',dict(name='bad',assetId=self.assets[self.bid]['id'])),
|
|
('/api/recorders',dict(name='bad',siteId=self.sites[self.bid]['id'])),
|
|
('/api/sites',dict(self.sites[self.aid],familyId=self.bid))]
|
|
for path,data in cases: self.assertIn(self.request(path,data=data)[0],(403,404))
|
|
self.assertEqual(self.request('/api/sites',data=dict(self.sites[self.aid],name='own edit'))[0],200)
|
|
|
|
def test_other_device_address_and_onvif_credentials_protected(self):
|
|
with patch.object(app.onvif,'inspect') as inspect:
|
|
for data in (dict(id=self.cameras[self.bid]['id']),dict(host=self.cameras[self.bid]['host'],username='admin')):
|
|
self.assertIn(self.request('/api/onvif',data=data)[0],(403,404))
|
|
inspect.assert_not_called()
|
|
self.assertEqual(self.request('/api/cameras',data=dict(name='duplicate',assetId=self.assets[self.aid]['id'],host=self.cameras[self.bid]['host']))[0],403)
|
|
|
|
def test_manager_cannot_escalate_or_touch_another_family_user(self):
|
|
for data in (dict(username='root2',password='fixture-only',role='admin'),
|
|
dict(self.other,name='stolen'), dict(self.root,name='hijack'),
|
|
dict(username='cross',password='fixture-only',siteIds=[self.sites[self.bid]['id']]),
|
|
dict(username='cross',password='fixture-only',relatedToId=self.other['id'])):
|
|
self.assertIn(self.request('/api/users',data=data)[0],(403,404))
|
|
|
|
def test_global_operations_require_super_admin(self):
|
|
for path,data in (('/api/storage',{}),('/api/discover',{}),('/api/households',{'name':'unauthorized'})):
|
|
self.assertEqual(self.request(path,data=data)[0],403)
|
|
self.assertEqual(self.request('/api/audit')[0],403)
|
|
self.assertEqual(self.request('/api/households',self.members[self.aid],{'name':'unauthorized'})[0],403)
|
|
|
|
def test_genealogy_revisions_history_restore_and_refs_isolated(self):
|
|
a = app.FAMILY.save('person',dict(name='A private',revision=0),self.owner)['item']
|
|
b = app.FAMILY.save('person',dict(name='B private',revision=0),self.other)['item']
|
|
self.assertEqual(self.request('/api/family')[1]['revision'],1)
|
|
self.assertEqual([p['name'] for p in self.request('/api/family')[1]['people']],['A private'])
|
|
revision=app.FAMILY.history(self.other,'person',b['id'])['items'][0]['revision']
|
|
for path in (f'/api/family/history?kind=person&id={b["id"]}', f'/api/family/change?revision={revision}'):
|
|
self.assertEqual(self.request(path)[0],404)
|
|
for path,data in (('/api/family/restore',dict(changeRevision=revision,side='after',revision=1)),
|
|
('/api/family/person',dict(b,revision=1)),
|
|
('/api/family/link',dict(fromId=a['id'],toId=b['id'],kind='parent',revision=1)),
|
|
('/api/family/event',dict(personIds=[b['id']],revision=1))):
|
|
self.assertIn(self.request(path,data=data)[0],(403,404))
|
|
self.assertEqual(self.request('/api/users',data=dict(username='crossperson',password='fixture-only',personId=b['id']))[0],404)
|
|
|
|
def test_leadership_backup_and_explicit_emergency_handover(self):
|
|
self.assertEqual(self.leadership()[0],200)
|
|
# Backup remains operational even if the owner's browser is unavailable.
|
|
self.assertEqual(self.request('/api/sites',self.backup,dict(name='emergency operation'))[0],200)
|
|
self.assertEqual(self.leadership(self.backup,revision=1,ownerId=self.backup['id'],successorId='')[0],403)
|
|
self.assertEqual(self.leadership(self.root,revision=1,ownerId=self.backup['id'],successorId=self.owner['id'],reason='家主无法操作，由超级管理员办理交接')[0],200)
|
|
home=app.get_object('households',self.aid)
|
|
self.assertEqual(home['ownerId'],self.backup['id']);self.assertEqual(len(home['leadershipHistory']),2)
|
|
self.assertEqual(app.get_object('users',self.owner['id'])['role'],'family_admin')
|
|
self.assertEqual(self.leadership(self.owner,revision=2)[0],403)
|
|
self.assertEqual(self.request('/api/households',self.members[self.aid])[1]['households'][0].get('leadershipHistory'),None)
|
|
|
|
def test_leadership_conflicts_and_account_lockout_prevention(self):
|
|
self.assertEqual(self.leadership()[0],200)
|
|
self.assertEqual(self.leadership()[0],409)
|
|
for uid in (self.owner['id'],self.backup['id']):
|
|
user=app.get_object('users',uid)
|
|
for fields in ({'disabled':True},{'role':'member'}):
|
|
self.assertEqual(self.request('/api/users',self.root,dict(user,**fields))[0],400)
|
|
self.assertEqual(self.request('/api/users',self.backup,dict(self.owner,password='hijack-password'))[0],403)
|
|
self.assertEqual(self.leadership(revision=1,successorId=self.other['id'])[0],404)
|
|
self.assertEqual(self.leadership(revision=1,successorId=self.owner['id'])[0],400)
|
|
self.assertEqual(self.leadership(revision=1,ownerId=self.members[self.aid]['id'])[0],400)
|
|
# Editing household metadata must preserve leadership and its history.
|
|
self.assertEqual(self.request('/api/households',data=dict(id=self.aid,name='renamed'))[0],200)
|
|
self.assertEqual(app.get_object('households',self.aid)['ownerId'],self.owner['id'])
|
|
|
|
def test_legacy_migration_idempotent_preserves_credentials(self):
|
|
user=app.get_object('users',self.owner['id']);user.pop('familyId')
|
|
app.save_object('users',user)
|
|
before=app.DB.execute('SELECT * FROM sessions ORDER BY hash').fetchall()
|
|
app.HOUSEHOLDS.initialize();app.HOUSEHOLDS.initialize()
|
|
saved=app.get_object('users',user['id']);self.assertEqual(saved.pop('familyId'),self.aid)
|
|
self.assertEqual(saved,user)
|
|
self.assertEqual([tuple(r) for r in before],[tuple(r) for r in app.DB.execute('SELECT * FROM sessions ORDER BY hash')])
|
|
|
|
|
|
if __name__=='__main__': unittest.main()
|