67 lines
3.2 KiB
Python
67 lines
3.2 KiB
Python
|
"""Environment credential precedence, rotation and account ownership boundaries."""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from pydantic import SecretStr, ValidationError
|
||
|
|
|
||
|
|
from app.config import Settings
|
||
|
|
from app.models import Account, Admin
|
||
|
|
from app.security import bootstrap, cipher
|
||
|
|
|
||
|
|
|
||
|
|
def test_dotenv_and_process_precedence(settings, tmp_path, monkeypatch):
|
||
|
|
env = tmp_path / '.env'
|
||
|
|
env.write_text("WQ_EMAIL=local@example.com\nWQ_PASSWORD='literal-$value#password'\n")
|
||
|
|
values = settings.model_dump(exclude={'wq_email', 'wq_password'})
|
||
|
|
local = Settings(_env_file=env, **values)
|
||
|
|
assert local.wq_email == 'local@example.com'
|
||
|
|
assert local.wq_password.get_secret_value() == 'literal-$value#password'
|
||
|
|
monkeypatch.setenv('WQ_EMAIL', 'production@example.com')
|
||
|
|
monkeypatch.setenv('WQ_PASSWORD', 'production-secret')
|
||
|
|
production = Settings(_env_file=env, **values)
|
||
|
|
assert production.wq_email == 'production@example.com'
|
||
|
|
assert production.wq_password.get_secret_value() == 'production-secret'
|
||
|
|
assert 'production-secret' not in repr(production)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('email,password', [('person@example.com', ''), ('', 'private-value')])
|
||
|
|
def test_partial_configuration_fails_without_leaking(settings, email, password):
|
||
|
|
values = settings.model_dump(exclude={'wq_email', 'wq_password'})
|
||
|
|
with pytest.raises(ValidationError) as error:
|
||
|
|
Settings(_env_file=None, **values, wq_email=email, wq_password=password)
|
||
|
|
assert 'must be configured together' in str(error.value)
|
||
|
|
assert 'private-value' not in str(error.value)
|
||
|
|
assert 'person@example.com' not in str(error.value)
|
||
|
|
|
||
|
|
|
||
|
|
async def test_env_rotation_api_lock_and_bound_account(app, logged_in):
|
||
|
|
settings = app.state.settings
|
||
|
|
settings.wq_email = 'person@example.com'
|
||
|
|
settings.wq_password = SecretStr('initial-env-secret')
|
||
|
|
async with app.state.sessions() as db:
|
||
|
|
original_admin = (await db.get(Admin, 1)).password_hash
|
||
|
|
await bootstrap(db, settings)
|
||
|
|
account = await db.get(Account, 1)
|
||
|
|
assert cipher(settings).decrypt(account.password_encrypted.encode()) == b'initial-env-secret'
|
||
|
|
account.wq_user_id = 'bound-user'
|
||
|
|
await db.commit()
|
||
|
|
settings.wq_password = SecretStr('rotated-env-secret')
|
||
|
|
async with app.state.sessions() as db:
|
||
|
|
await bootstrap(db, settings)
|
||
|
|
account = await db.get(Account, 1)
|
||
|
|
assert cipher(settings).decrypt(account.password_encrypted.encode()) == b'rotated-env-secret'
|
||
|
|
assert (await db.get(Admin, 1)).password_hash == original_admin
|
||
|
|
response = await logged_in.get('/api/v1/account')
|
||
|
|
assert response.json()['credentials_source'] == 'environment'
|
||
|
|
assert response.json()['configured'] is True
|
||
|
|
assert 'secret' not in response.text and 'password' not in response.text
|
||
|
|
response = await logged_in.put('/api/v1/account/credentials', json={
|
||
|
|
'email': 'person@example.com', 'password': 'manual-secret',
|
||
|
|
})
|
||
|
|
assert response.status_code == 409
|
||
|
|
settings.wq_email = 'other@example.com'
|
||
|
|
async with app.state.sessions() as db:
|
||
|
|
with pytest.raises(ValueError, match='bound WorldQuant account'):
|
||
|
|
await bootstrap(db, settings)
|
||
|
|
await db.rollback()
|
||
|
|
assert (await db.get(Account, 1)).email == 'person@example.com'
|