125 lines
4.8 KiB
Python
125 lines
4.8 KiB
Python
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui'))
|
|
import server
|
|
|
|
|
|
def setUpModule():
|
|
__import__('importlib').reload(server)
|
|
|
|
|
|
class SessionTest(unittest.TestCase):
|
|
def setUp(self):
|
|
fd, self.path = tempfile.mkstemp()
|
|
os.close(fd)
|
|
server.SESSION_FILE = self.path
|
|
|
|
def tearDown(self):
|
|
try:
|
|
os.unlink(self.path)
|
|
except OSError:
|
|
pass
|
|
|
|
def test_save_and_load_roundtrip(self):
|
|
server.save_session({'serverid': '001337AEE050', 'token': 'abc123'})
|
|
self.assertEqual(server.load_session(), {'serverid': '001337AEE050', 'token': 'abc123'})
|
|
if os.name != 'nt':
|
|
self.assertEqual(stat.S_IMODE(os.stat(self.path).st_mode), 0o600)
|
|
|
|
def test_load_missing_file_returns_empty(self):
|
|
os.unlink(self.path)
|
|
self.assertEqual(server.load_session(), {})
|
|
|
|
def test_check_auth_matches_cookie(self):
|
|
server.save_session({'serverid': '001337AEE050', 'token': 'abc123'})
|
|
self.assertTrue(server.check_auth('foo=bar; AUTH_001337AEE050=abc123'))
|
|
self.assertFalse(server.check_auth('AUTH_001337AEE050=wrong'))
|
|
self.assertFalse(server.check_auth(''))
|
|
|
|
def test_check_auth_rejects_prefix_value(self):
|
|
server.save_session({'serverid': '001337AEE050', 'token': 'abc123'})
|
|
self.assertFalse(server.check_auth('AUTH_001337AEE050=abc123xxxx'))
|
|
self.assertFalse(server.check_auth('prefix AUTH_001337AEE050=abc123'))
|
|
server.SESSION_FILE = os.path.join(tempfile.gettempdir(), 'pagerwebui_test_session.json')
|
|
if os.path.exists(server.SESSION_FILE):
|
|
os.unlink(server.SESSION_FILE)
|
|
self.assertFalse(server.check_auth('AUTH_=anything'))
|
|
|
|
def test_current_token(self):
|
|
server.save_session({'serverid': 'S', 'token': 'tok'})
|
|
self.assertEqual(server.current_token(), 'tok')
|
|
|
|
def test_logout_removes_session_and_expires_cookie(self):
|
|
server.save_session({'serverid': 'S', 'token': 'tok'})
|
|
class H:
|
|
def __init__(self):
|
|
self.headers = []
|
|
def add_extra_header(self, name, value):
|
|
self.headers.append((name, value))
|
|
ctx = type('C', (), {'h': H()})()
|
|
status, payload = server.h_logout(ctx)
|
|
self.assertEqual(status, 200)
|
|
self.assertTrue(payload['ok'])
|
|
self.assertEqual(server.load_session(), {})
|
|
self.assertIn(('Set-Cookie',
|
|
'AUTH_S=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0'),
|
|
ctx.h.headers)
|
|
|
|
|
|
class LoginTest(unittest.TestCase):
|
|
def test_login_success_sets_cookie_payload(self):
|
|
def fake_daemon(method, path, body=None, token=None, timeout=15):
|
|
if path == '/api/login':
|
|
return 200, {'token': 'tok123'}
|
|
if path == '/api/api_ping':
|
|
return 200, {'serverid': '001337AEE050', 'version': '1.0'}
|
|
return 404, None
|
|
server.daemon_call = fake_daemon
|
|
server.SESSION_FILE = os.path.join(tempfile.gettempdir(), 'pagerwebui_test_session.json')
|
|
if os.path.exists(server.SESSION_FILE):
|
|
os.unlink(server.SESSION_FILE)
|
|
class H:
|
|
def __init__(self):
|
|
self.headers = []
|
|
def add_extra_header(self, name, value):
|
|
self.headers.append((name, value))
|
|
class Ctx:
|
|
body = {'username': 'root', 'password': 'pw'}
|
|
cookie = ''
|
|
h = H()
|
|
ctx = Ctx()
|
|
status, payload = server.h_login(ctx)
|
|
self.assertEqual(status, 200)
|
|
self.assertNotIn('token', payload)
|
|
self.assertTrue(payload['ok'])
|
|
self.assertEqual(payload['serverid'], '001337AEE050')
|
|
self.assertEqual(server.load_session()['token'], 'tok123')
|
|
self.assertTrue(any(n == 'Set-Cookie' and v.startswith('AUTH_001337AEE050=tok123') for n, v in ctx.h.headers))
|
|
|
|
def test_login_failure_returns_401(self):
|
|
def fake_daemon(method, path, body=None, token=None, timeout=15):
|
|
return 401, {'error': 'bad'}
|
|
server.daemon_call = fake_daemon
|
|
class Ctx:
|
|
body = {'username': 'root', 'password': 'nope'}
|
|
cookie = ''
|
|
status, payload = server.h_login(Ctx())
|
|
self.assertEqual(status, 401)
|
|
|
|
def test_api_ping(self):
|
|
server.daemon_call = lambda *a, **k: (200, {'serverid': 'S1', 'version': '9'})
|
|
server.SESSION_FILE = os.path.join(tempfile.gettempdir(), 'pagerwebui_test_session.json')
|
|
status, payload = server.h_api_ping(type('C', (), {'cookie': '', 'args': ()})())
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(payload['serverid'], 'S1')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|