Files
dtf-system/local/security_test.py
Cauê Faleiros c1a07a75fa
All checks were successful
Build and deploy / Validate source (push) Successful in 4s
Build and deploy / Integration suite on a real stack (push) Successful in 1m12s
Build and deploy / Secret scan and release gate (push) Successful in 5s
Build and deploy / Publish images and notify Portainer (push) Successful in 1m50s
ci: run the integration suites inside the stack's own network
The suites connected to localhost:<published port>, which works for a developer
but not on a containerised runner: published ports live in the host's network
namespace, so the runner container gets connection refused.

Run them from inside the stack instead, against the gateway by service name.
SITE_BASE_URL and SITE_HOST_HEADER make that possible without weakening what is
under test: the Host stays "localhost", so the gateway's host check and
TrustedHostMiddleware see exactly what a localhost run produces, and the tests
that deliberately send their own Host still override it.

S3_PUBLIC_ENDPOINT has to agree, because presigned URLs are signed against it
and the signature covers the host, so it cannot be rewritten afterwards. CI
points the whole stack at http://storage:9000 so the URLs it hands out are
reachable by whoever follows them.

The browser suites still need Chrome to reach the stack from the runner, which
the same namespace split prevents. They now check reachability and skip with a
warning instead of failing with a bare connection error; recorded as ROADMAP
5.10, since they are the only coverage for the artwork editor.

Verified both ways: the six suites pass inside the network, and an unchanged
developer localhost run still passes, as do both browser suites locally.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-21 13:29:15 -03:00

78 lines
4.2 KiB
Python

"""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 .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
assert "'sha256-" in policy and "object-src 'none'" in policy
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()