cms-SiteForge/tests/test_p1_platform.py

"""Deterministic P1 tests: no actual network or external infrastructure."""
import socket
from app import safe_http, secret_refs
from fastapi.testclient import TestClient
from app import db
from app.main import app
import pytest


def test_public_ip_pinned_and_bounded(monkeypatch):
    seen=[]
    def dns(host, port, **kwargs):
        return [(socket.AF_INET,socket.SOCK_STREAM,6,'',('93.184.216.34',port))]
    monkeypatch.setattr(safe_http.socket,'getaddrinfo',dns)
    class Response:
        status=200
        def getheaders(self):return [('Content-Type','text/html')]
        def read(self,n):return b'hello'
    class FakeConnection:
        def __init__(self,host,ip,port,timeout):seen.append((host,ip,port,timeout))
        def request(self,method,url,headers,body=None):seen.append((method,url))
        def getresponse(self):return Response()
        def close(self):pass
    monkeypatch.setattr(safe_http,'_PinnedHTTPS',FakeConnection)
    r=safe_http.fetch('https://example.org/products/?page=1',max_bytes=8)
    assert r.content==b'hello'
    assert seen[0]==('example.org','93.184.216.34',443,12)
    assert seen[1]==('GET','/products/?page=1')


def test_public_only_dns_and_url_restrictions(monkeypatch):
    def dns(host,port,**kwargs):
        ip={'rebinding.example':'10.0.0.5','mixed.example':'93.184.216.34'}.get(host,'93.184.216.34')
        r=[(socket.AF_INET,socket.SOCK_STREAM,6,'',(ip,port))]
        if host=='mixed.example':r.append((socket.AF_INET,socket.SOCK_STREAM,6,'',('127.0.0.1',port)))
        return r
    monkeypatch.setattr(safe_http.socket,'getaddrinfo',dns)
    for url in ['http://localhost/','http://169.254.169.254/latest/meta-data/',
                'https://rebinding.example/','https://mixed.example/',
                'file:///etc/passwd','http://user:pass@example.org/',
                'https://example.org:8080/']:
        with pytest.raises(safe_http.OutboundError):safe_http.resolve_target(url)
    assert safe_http.resolve_target('https://example.org/')[3]=='93.184.216.34'


def test_redirect_to_private_host_is_not_followed(monkeypatch):
    def dns(host,port,**kwargs):
        ip='10.0.0.2' if host=='internal.example' else '93.184.216.34'
        return [(socket.AF_INET,socket.SOCK_STREAM,6,'',(ip,port))]
    monkeypatch.setattr(safe_http.socket,'getaddrinfo',dns)
    class Response:
        status=302
        def getheaders(self):return [('Location','http://internal.example/secret')]
        def read(self,n):return b''
    class FakeConnection:
        def __init__(self,*args):pass
        def request(self,*args,**kwargs):pass
        def getresponse(self):return Response()
        def close(self):pass
    monkeypatch.setattr(safe_http,'_PinnedHTTPS',FakeConnection)
    with pytest.raises(safe_http.OutboundError):safe_http.fetch('https://example.org/',redirects=3)


def test_secret_refs_deny_literals_and_path_traversal(monkeypatch):
    monkeypatch.setenv('SITEFORGE_UNIT_TOKEN','sample-secret')
    assert secret_refs.resolve('env:SITEFORGE_UNIT_TOKEN')=='sample-secret'
    for ref in ['my-plain-password','env:PATH','file:../root','file:/etc/passwd','env:SITEFORGE_NOT_SET']:
        with pytest.raises(secret_refs.SecretReferenceError):secret_refs.resolve(ref)


def test_queue_limits_and_repeated_requests(monkeypatch,tmp_path):
    monkeypatch.setattr(db,'DATA',tmp_path)
    monkeypatch.setattr(db,'DB_FILE',tmp_path/'p1.sqlite3')
    monkeypatch.setenv('SITEFORGE_MAX_PENDING_JOBS_PER_SITE','1')
    monkeypatch.setenv('SITEFORGE_MAX_PENDING_JOBS','10')
    monkeypatch.setenv('SITEFORGE_ADMIN_PASSWORD','p1-test-password-with-strong-entropy')
    monkeypatch.setenv('SITEFORGE_SESSION_SECRET','p1-test-secret-with-enough-entropy-1234567890')
    with TestClient(app) as client:
        assert client.post('/api/auth/login',json={'password':'p1-test-password-with-strong-entropy'}).status_code==200
        client.headers['X-SiteForge-CSRF']=client.get('/api/auth/me').json()['csrf']
        sid=client.post('/api/sites',json={'name':'Queue capacity'}).json()['id']
        one=client.post(f'/api/sites/{sid}/builds',json={'mirror_images':False})
        assert one.status_code==202
        duplicated=client.post(f'/api/sites/{sid}/builds',json={'mirror_images':False})
        assert duplicated.status_code==202 and duplicated.json()['existing']
        over=client.post(f'/api/sites/{sid}/builds',json={'mirror_images':True})
        assert over.status_code==429
        health=client.get('/api/health')
        assert health.status_code==200 and len(health.headers['X-Request-ID'])==16


def test_schema_ledger_idempotent_and_checksum_guard(monkeypatch,tmp_path):
    monkeypatch.setattr(db,'DATA',tmp_path)
    monkeypatch.setattr(db,'DB_FILE',tmp_path/'schema.sqlite3')
    db.init()
    db.init()
    assert db.one('SELECT COUNT(*) AS n FROM schema_migrations')['n']==1
    assert db.one("SELECT name FROM sqlite_master WHERE name='platform_limit_overrides'")
    db.execute("UPDATE schema_migrations SET checksum='corrupted'")
    with pytest.raises(RuntimeError,match='checksum mismatch'):db.init()


def test_post_uses_pinned_ip_and_never_redirects(monkeypatch):
    monkeypatch.setattr(safe_http.socket,'getaddrinfo',lambda host,port,**k:[
        (socket.AF_INET,socket.SOCK_STREAM,6,'',('93.184.216.34',port))])
    seen=[]
    class Response:
        status=201
        def getheaders(self):return [('Content-Type','application/json')]
        def read(self,n):return b'{"release":"xyz"}'
    class Connection:
        def __init__(self,host,ip,port,timeout):seen.append((host,ip))
        def request(self,method,path,body=None,headers=None):seen.append((method,path,body))
        def getresponse(self):return Response()
        def close(self):pass
    monkeypatch.setattr(safe_http,'_PinnedHTTPS',Connection)
    result=safe_http.post('https://releases.example.org/deploy/site1',body=b'zip-data')
    assert result.status_code==201 and result.json()['release']=='xyz'
    assert seen[0]==('releases.example.org','93.184.216.34')
    assert seen[1]==('POST','/deploy/site1',b'zip-data')