"""Non-destructive localhost security regressions. Leaves tiny test upload reservations."""
import base64
import os
from urllib.error import HTTPError
from urllib.request import Request, urlopen
from urllib.parse import urlparse, parse_qs
from uuid import uuid4
from tests.smoke_test import Client, BASE, with_host

def raw(path, expected, headers=None, body=None):
    request=Request(BASE+path, data=body, headers=with_host(headers))
    try:
        with urlopen(request,timeout=10) as response:
            assert response.status==expected
            return response.headers
    except HTTPError as error:
        assert error.code==expected,(path,error.code,expected)
        return error.headers

def run():
    headers=raw('/',200)
    policy=headers['Content-Security-Policy']
    assert "script-src-attr 'none'" in policy and "frame-ancestors 'none'" in policy
    # The Site has no inline script, so the policy needs no hash allowlist at all.
    # Assert the property that matters rather than the mechanism: nothing inline
    # executes, and any hash that does appear was added deliberately at build time.
    script_src = next(d.strip() for d in policy.split(';') if d.strip().startswith('script-src '))
    assert "'unsafe-inline'" not in script_src and "'unsafe-eval'" not in script_src, script_src
    assert script_src == "script-src 'self'", script_src
    assert "object-src 'none'" in policy
    assert 'cdnjs' not in policy, 'pdf.js is vendored; no CDN belongs in the policy'
    # Product pages and the cart are addresses of the Site's page, under the same policy.
    for page in ('/arquivo-por-metro','/artes-avulsas','/uv-arquivo-por-metro','/uv-artes-avulsas','/carrinho'):
        assert raw(page,200)['Content-Security-Policy']==policy,page
    raw('/carrinho/outra-coisa',404)
    raw('/api/health',400,{'Host':'attacker.invalid'})
    raw('/api/account/logout',403,{'Content-Type':'application/json','Origin':'https://attacker.invalid'},b'{}')
    raw('/api/account/logout',403,{'Content-Type':'application/json','Origin':'http://localhost:9999'},b'{}')
    print('PASS: CSP, frame protection, Host and cross-origin rejection')

    operator=Client()
    credentials={'email':os.environ.get('OPERATOR_EMAIL','operator@example.test'),'password':os.environ.get('OPERATOR_PASSWORD','local-operator-only')}
    encoded=base64.b64encode((credentials['email']+':'+credentials['password']).encode()).decode()
    raw('/api/operator/board',401,{'Authorization':'Basic '+encoded})
    operator.call('/operator/login',credentials)
    token=next(c for c in operator.jar if c.name=='dtf_operator')
    assert token.has_nonstandard_attr('HttpOnly') and token.get_nonstandard_attr('SameSite')=='strict'
    assert token.path=='/api/operator'
    operator.call('/operator/board')
    replay=Client();replay.jar.set_cookie(token)
    operator.call('/operator/logout',{})
    replay.call('/operator/board',expected=401)
    print('PASS: Basic rejected; HttpOnly scoped operator session; server-side logout revocation')

    client=Client();client.call('/session')
    client.call('/uploads',{'name':'payload.html','size':1},expected=422)
    uid=client.call('/uploads',{'name':'SECURITY-PART.cdr','size':3})['id']
    url=client.call('/uploads/'+uid+'/parts/1',{})['url']
    assert 'content-length' in parse_qs(urlparse(url).query)['X-Amz-SignedHeaders'][0]
    try:
        urlopen(Request(url,data=b'toolong',method='PUT'),timeout=10)
        raise AssertionError('Signed part accepted wrong length')
    except HTTPError as error:assert error.code==403,error.code
    with urlopen(Request(url,data=b'abc',method='PUT'),timeout=10) as response:assert response.status==200
    client.call('/uploads/'+uid+'/complete',{})
    count=int(os.environ.get('MAX_PENDING_UPLOADS','10'))
    for i in range(count):client.call('/uploads',{'name':'SECURITY-PENDING.cdr','size':1})
    client.call('/uploads',{'name':'SECURITY-OVER-LIMIT.cdr','size':1},expected=429)
    print('PASS: extension allowlist, exact multipart Content-Length signature, pending upload quota')

    # Unique identity avoids locking out the real local operator.
    attacker=Client();email='test-'+uuid4().hex+'@example.test'
    for _ in range(10):attacker.call('/operator/login',{'email':email,'password':'invalid'},expected=401)
    attacker.call('/operator/login',{'email':email,'password':'invalid'},expected=429)
    print('PASS: operator login throttling (only synthetic account bucket exhausted)')

    # Guest sessions are limited per source, not once for the whole deployment.
    # Keyed on the environment name this was a single global bucket of 120 per
    # 15 minutes, which the suites above would already have eaten into.
    for _ in range(25):
        Client().call('/session')
    # A forged forwarded address must not let a client pick another bucket: the
    # gateway overwrites the header, so these count against the real source too.
    for _ in range(5):
        raw('/api/session',200,{'X-Forwarded-For':'203.0.113.7'})
    print('PASS: guest sessions limited per source, forwarded address not client-controlled')

if __name__=='__main__':run()