dtf-system/tests/runtime_security_test.py

"""Run inside API container: permissions, hashing and fail-closed scanner unit checks."""
import hashlib
from unittest.mock import patch
from botocore.exceptions import ClientError
from app.core.db import connect
from app.adapters import LocalS3Storage
from app.core.auth import client_ip, password_hash, password_matches
from app.scanning import ClamAV, format_matches, require_clean
from fastapi import HTTPException

class FakeRequest:
    def __init__(self, headers=None, peer='10.0.0.2'):
        self.headers=headers or {}
        self.client=type('C',(),{'host':peer})() if peer else None

def check_client_ip():
    """The gateway hands one address; a direct peer falls back to its own."""
    # Gateway-set header wins over the connection peer, which is the gateway.
    assert client_ip(FakeRequest({'x-forwarded-for':'198.51.100.9'}))=='198.51.100.9'
    # Only the first entry is used, and it is length-capped.
    assert client_ip(FakeRequest({'x-forwarded-for':'198.51.100.9, 10.0.0.2'}))=='198.51.100.9'
    assert len(client_ip(FakeRequest({'x-forwarded-for':'a'*500})))<=64
    # No header: the peer address, never a constant shared by every request.
    assert client_ip(FakeRequest(peer='192.0.2.5'))=='192.0.2.5'
    assert client_ip(FakeRequest({'x-forwarded-for':'   '},peer='192.0.2.5'))=='192.0.2.5'
    assert client_ip(FakeRequest(peer=None))=='unknown'
    print('PASS: client address resolution for rate-limit buckets and audit events')

def check_operator_accounts():
    """Accounts are per person, and disabling one ends its access at once."""
    from uuid import uuid4
    from app.core.auth import password_hash, password_matches
    from app.operators import seed_from_environment, set_active
    email='runtime-check-'+uuid4().hex[:8]+'@example.test'
    with connect() as c:
        c.execute('INSERT INTO dtf_local.operators(id,email,name,password_hash) VALUES(%s,%s,%s,%s)',
                  (uuid4(), email, 'Runtime Check', password_hash('runtime-check-password')))
        row=c.execute('SELECT * FROM dtf_local.operators WHERE email=%s',(email,)).fetchone()
        assert row['active'] and password_matches('runtime-check-password', row['password_hash'])
        assert not password_matches('wrong', row['password_hash'])
        # Seeding is once-only: a password changed here must survive a redeploy.
        c.execute("INSERT INTO dtf_local.operator_sessions(token_hash,username) VALUES(%s,%s)",
                  ('runtime-check-'+uuid4().hex, email))
    set_active(email, False)
    with connect() as c:
        row=c.execute('SELECT active FROM dtf_local.operators WHERE email=%s',(email,)).fetchone()
        sessions=c.execute('SELECT count(*) AS n FROM dtf_local.operator_sessions WHERE username=%s',
                           (email,)).fetchone()['n']
        assert not row['active'], 'disable did not deactivate'
        assert sessions==0, 'disable left an open session behind'
        c.execute('DELETE FROM dtf_local.operators WHERE email=%s',(email,))
    print('PASS: per-operator accounts, password verification, immediate revocation on disable')

def run():
    check_client_ip()
    check_operator_accounts()
    with connect() as c:
        role=c.execute('SELECT rolsuper,rolcreatedb,rolcreaterole,rolbypassrls FROM pg_roles WHERE rolname=current_user').fetchone()
        assert not any(role.values()),role
        assert not c.execute("SELECT has_schema_privilege(current_user,'dtf_local','CREATE') AS allowed").fetchone()['allowed']
    storage=LocalS3Storage()
    storage.health()
    visible={bucket['Name'] for bucket in storage.client.list_buckets()['Buckets']}
    assert visible=={storage.bucket},visible
    for call in (lambda:storage.client.get_bucket_policy(Bucket=storage.bucket),
                 lambda:storage.client.get_object(Bucket=storage.bucket,Key='outside-runtime-prefix/test.cdr')):
        try:call();raise AssertionError('Runtime storage credentials have excessive privilege')
        except ClientError as error:assert error.response['ResponseMetadata']['HTTPStatusCode']==403
    password='test-password-for-hash'
    salt='00'*16
    old='scrypt$'+salt+'$'+hashlib.scrypt(password.encode(),salt=bytes.fromhex(salt),n=16384,r=8,p=1).hex()
    assert password_matches(password,old)
    new=password_hash(password)
    assert new.startswith('scrypt-v2$') and password_matches(password,new)
    assert not password_matches('wrong',new)
    for state in ('pending','error','rejected'):
        try:require_clean({'complete':True,'scan_state':state});raise AssertionError('Unscanned file released')
        except HTTPException as error:assert error.status_code==409
    require_clean({'complete':True,'scan_state':'clean'})
    assert ClamAV().ping() and ClamAV().version().startswith('ClamAV ')
    # Above the antivirus limit only a file whose bytes match its name is released.
    for name,head,ok in (('folha.png',b'\x89PNG\r\n\x1a\n\x00',True),('folha.png',b'MZ\x90\x00',False),
                         ('folha.jpg',b'\xff\xd8\xff\xe0',True),('folha.pdf',b'%PDF-1.7',True),
                         ('folha.pdf',b'#!/bin/sh',False),('folha.tif',b'II*\x00',True),('folha.psd',b'8BPS',True),
                         ('folha.ai',b'%!PS-Adobe',True),('folha.cdr',b'RIFF\x10\x00\x00\x00CDRv',True),
                         ('folha.cdr',b'RIFF\x10\x00\x00\x00WEBP',False),('folha.exe',b'MZ',False)):
        assert format_matches(name,head)==ok,(name,head)
    with patch('app.scanning.socket.create_connection',side_effect=OSError('offline')):
        try:ClamAV().scan(None,1);raise AssertionError('Offline scanner returned success')
        except OSError:pass
    print('PASS: runtime DB/S3 least privilege, legacy/current password hashes, quarantine states and scanner size/offline behavior')

if __name__=='__main__':run()