101 lines
5.3 KiB
Python
101 lines
5.3 KiB
Python
import json
|
|
import sys
|
|
from types import SimpleNamespace
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
from viano_bug_analysis.api import APIConfig,Project,APIError
|
|
from viano_bug_analysis.api.app import create_app
|
|
from viano_bug_analysis.api.users import Users
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_app(tmp_path):
|
|
config=APIConfig(tmp_path/'service',(Project('p','wow','vcore'),Project('second','wow','vcore')),())
|
|
app=create_app(config)
|
|
app.state.sessions.users.manage('create_user','alice','first password')
|
|
app.state.sessions.users.manage('create_user','bob','other password')
|
|
return app
|
|
|
|
|
|
def login(client,name='alice',password='first password'):
|
|
response=client.post('/v1/auth/login',json={'username':name,'password':password})
|
|
assert response.status_code==200,response.text
|
|
return {'X-CSRF-Token':response.json()['csrf']}
|
|
|
|
|
|
def test_login_csrf_permissions_password_change_and_delete(auth_app):
|
|
client=TestClient(auth_app)
|
|
assert client.post('/v1/auth/login',json={'username':'alice','password':'wrong'}).status_code==401
|
|
assert client.post('/v1/auth/login',headers={'Origin':'https://evil.example'},json={'username':'alice','password':'first password'}).status_code==403
|
|
auth=login(client)
|
|
assert 'httponly' in client.post('/v1/auth/login',json={'username':'alice','password':'first password'}).headers['set-cookie'].lower()
|
|
auth={'X-CSRF-Token':client.get('/v1/auth/session').json()['csrf']}
|
|
assert {p['project_id'] for p in client.get('/v1/profiles').json()['projects']}=={'p','second'}
|
|
assert client.get('/v1/operations').json()['operator']
|
|
command={'action':'pause','project_id':'p'}
|
|
assert client.post('/v1/operations',json=command).status_code==403
|
|
assert client.post('/v1/operations',headers=auth,json=command).status_code==200
|
|
with pytest.raises(APIError):auth_app.state.sessions.users.manage('change_pw_user','alice','new password','wrong')
|
|
auth_app.state.sessions.users.manage('change_pw_user','alice','new password','first password')
|
|
assert client.get('/v1/profiles').status_code==401
|
|
auth=login(client,password='new password')
|
|
auth_app.state.sessions.users.manage('delete_user','alice')
|
|
assert client.get('/v1/auth/session').status_code==401
|
|
|
|
|
|
def test_claim_is_atomic_session_bound_and_released_on_logout_or_expiry(auth_app):
|
|
sessions=auth_app.state.sessions
|
|
now=[1000.0];sessions.clock=lambda:now[0]
|
|
first,second=TestClient(auth_app),TestClient(auth_app)
|
|
first_auth,second_auth=login(first),login(second,'bob','other password')
|
|
def sid(client):return sessions.digest(client.cookies['vba_session'])
|
|
identifiers=[sid(first),sid(second)]
|
|
def acquire(identifier):
|
|
try:sessions.claim(identifier,'case','request',1);return 'won'
|
|
except APIError as error:return error.code
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results=list(pool.map(acquire,identifiers))
|
|
assert sorted(results)==['API_TEST_CLAIMED','won']
|
|
winner=results.index('won');loser=1-winner
|
|
clients=[first,second];headers=[first_auth,second_auth]
|
|
assert clients[winner].post('/v1/auth/logout',headers=headers[winner]).status_code==200
|
|
sessions.claim(identifiers[loser],'case','request',1)
|
|
sessions.claim(identifiers[loser],'case','request',1,check=True)
|
|
now[0]+=auth_app.state.sessions.config.session_seconds+1
|
|
assert clients[loser].get('/v1/auth/session').status_code==401
|
|
login(first)
|
|
sessions.claim(sid(first),'case','request',1)
|
|
|
|
|
|
def test_login_rate_limit_persists_across_requests(auth_app):
|
|
client=TestClient(auth_app)
|
|
for _ in range(10):
|
|
assert client.post('/v1/auth/login',json={'username':'missing','password':'wrong'}).status_code==401
|
|
assert client.post('/v1/auth/login',json={'username':'missing','password':'wrong'}).status_code==429
|
|
|
|
|
|
def test_console_privilege_guard_and_commands(tmp_path,monkeypatch,capsys):
|
|
from viano_bug_analysis.api.__main__ import main
|
|
import viano_bug_analysis.api.users as module
|
|
import viano_bug_analysis.api.__main__ as cli
|
|
configuration=tmp_path/'api.toml'
|
|
configuration.write_text('[api]\nworkspace="service"\n[[projects]]\nid="p"\ntype="wow"\nprofile="vcore"\n')
|
|
monkeypatch.setattr(module,'os',SimpleNamespace(name='posix',geteuid=lambda:1000))
|
|
monkeypatch.setattr(sys,'argv',['api','create_user','alice','private-password','--config',str(configuration)])
|
|
assert main()==1
|
|
output=capsys.readouterr().out
|
|
assert 'AUTH_PRIVILEGED_REQUIRED' in output and 'private-password' not in output
|
|
monkeypatch.undo()
|
|
monkeypatch.setattr(module,'require_privileged',lambda:None)
|
|
if hasattr(cli.os,'chown'):monkeypatch.setattr(cli.os,'chown',lambda *args:None)
|
|
monkeypatch.setattr(sys,'argv',['api','create_user','alice','private-password','--config',str(configuration)])
|
|
assert main()==0
|
|
users=Users(tmp_path/'service/auth/users.sqlite3')
|
|
assert users.verify('alice','private-password')
|
|
monkeypatch.setattr(sys,'argv',['api','change_pw_user','alice','private-password','new-password','--config',str(configuration)])
|
|
assert main()==0 and users.verify('alice','new-password')
|
|
monkeypatch.setattr(sys,'argv',['api','delete_user','alice','--config',str(configuration)])
|
|
assert main()==0 and users.get('alice') is None
|