Spaces:
Sleeping
Sleeping
File size: 3,670 Bytes
8f1b6ce | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | import asyncio
import ipaddress
from typing import Any
from unittest.mock import patch
import httpcore
import pytest
from conftest import register
from cryptography.fernet import Fernet, InvalidToken
from fastapi import HTTPException
from app.config import Settings
from app.security import SafeNetworkBackend, Vault, validate_base_url, validate_headers
from app.state import SharedState
@pytest.mark.parametrize(
"url",
[
"file:///etc/passwd",
"http://example.com/v1",
"https://name:secret@example.com/v1",
"https://example.com?key=secret",
"https://example.com/#a",
"https://example.com\\@127.0.0.1",
"https://example.com:invalid",
"https://example.com/\n",
],
)
def test_reject_invalid_upstreams(url: str) -> None:
with pytest.raises(HTTPException):
validate_base_url(url, [])
@pytest.mark.parametrize(
"address",
["127.0.0.1", "169.254.169.254", "10.0.0.1", "192.168.0.1", "::1", "::ffff:127.0.0.1"],
)
async def test_transport_blocks_private_dns_at_connect(address: str) -> None:
family = 10 if ipaddress.ip_address(address).version == 6 else 2
async def resolve(*args: Any, **kwargs: Any) -> list:
return [(family, 1, 6, "", (address, 443))]
with patch.object(asyncio.get_running_loop(), "getaddrinfo", resolve):
with pytest.raises(httpcore.ConnectError, match="blocked"):
await SafeNetworkBackend([]).connect_tcp("public-looking.example", 443)
async def test_mixed_dns_answers_are_rejected(environment: Any) -> None:
async def resolve(*args: Any, **kwargs: Any) -> list:
return [(2, 1, 6, "", ("8.8.8.8", 443)), (2, 1, 6, "", ("127.0.0.1", 443))]
with patch.object(asyncio.get_running_loop(), "getaddrinfo", resolve):
with pytest.raises(httpcore.ConnectError):
await SafeNetworkBackend([]).connect_tcp("rebind.example", 443)
@pytest.mark.parametrize(
"headers",
[
{"Host": "internal"},
{"Cookie": "secret"},
{"X-Test": "ok\r\nHost: internal"},
{"Proxy-Authorization": "secret"},
{"Content-Length": "1"},
],
)
def test_forbidden_custom_headers(headers: dict[str, str]) -> None:
with pytest.raises(HTTPException):
validate_headers(headers)
def test_key_rotation_and_integrity() -> None:
old, new = Fernet.generate_key().decode(), Fernet.generate_key().decode()
sealed = Vault(old).seal({"api_key": "private"})
rotating = Vault(new + "," + old)
assert rotating.open(sealed)["api_key"] == "private"
rotated = rotating.cipher.rotate(sealed.encode()).decode()
assert Vault(new).open(rotated)["api_key"] == "private"
with pytest.raises(InvalidToken):
Vault(old).open(rotated)
async def test_concurrent_rate_limit_atomicity() -> None:
shared = SharedState(None)
async def call() -> bool:
try:
await shared.rate_limit("test", 5)
return True
except HTTPException:
return False
assert sum(await asyncio.gather(*(call() for _ in range(20)))) == 5
def test_production_refuses_insecure_configuration() -> None:
with pytest.raises(ValueError, match="PostgreSQL"):
Settings(encryption_keys=Fernet.generate_key().decode(), environment="production")
async def test_oversized_and_chunked_body(environment: Any) -> None:
app, client, _ = environment
await register(client)
app.state.settings.max_request_bytes = 1024
async def body() -> Any:
yield b"x" * 700
yield b"x" * 700
result = await client.post("/api/providers", content=body())
assert result.status_code == 413
|