badblocks/docker-easy-haproxy
1
0
Fork 0
Code Activity

Refactor Docker Compose examples and plugins, add pytest tests

Browse source 
- Adjusted Compose files to build images locally instead of pulling.
- Simplified examples by removing `/etc/hosts` entry instructions.
- Enhanced `cloudflare.py` plugin with a configurable log format.
- Improved `generate-keys.sh` for portability using `$SCRIPT_DIR`.
- Added end-to-end tests for Compose examples using pytest.
This commit is contained in:
Joao Gilberto Magalhaes 2026-02-11 19:19:08 -05:00
parent 343987ab58
commit 5767e55dea
19 changed files with 1162 additions and 89 deletions
Download patch file Download diff file Expand all files Collapse all files

18
examples/docker/AGENTS.md Normal file
View file

@ -0,0 +1,18 @@
# Instructions for testing
1. Run a docker compose in background for the specified feature e.g. `docker compose -f docker-compose.yml up -d`
2. Check if it is running by running `docker ps` and verifying the container is up
3. If the container is not running, check the logs with `docker logs <container_id>` to diagnose any issues
4. In the top each file, you can find the instructions to test and check if it is working.
5. If everything is working tear down the container with `docker compose -f docker-compose.yml down`
6. To ensure the container is properly shut down, use `docker compose -f docker-compose.yml down --remove-orphans` to remove any orphaned containers.
# In case you find issues
**DONT TEAR DOWN THE CONTAINERS**
1. Investigate the source code in src/*
2. Try to fix it.
3. After the code is changed, build it again: `docker build -t byjg/easy-haproxy:5.0.0 -f build/Dockerfile --no-cache .` and start the tests again.

10
examples/docker/docker-compose-changed-label.yml
View file

@ -7,11 +7,6 @@
# - Useful for running multiple EasyHAProxy instances
# - Custom label configuration (haproxy.* instead of easyhaproxy.*)
#
# REQUIREMENTS (run these first):
# ```bash
# # Add to /etc/hosts (idempotent)
# grep -q "host1.local" /etc/hosts || echo "127.0.0.1 host1.local" | sudo tee -a /etc/hosts
# ```
#
# HOW TO START:
# ```bash
@ -38,7 +33,10 @@
services:
haproxy:
image: byjg/easy-haproxy:5.0.0
build:
context: ../../
dockerfile: build/Dockerfile
image: byjg/easy-haproxy:local
volumes:
- /var/run/docker.sock:/var/run/docker.sock
environment:

14
examples/docker/docker-compose-cloudflare.yml
View file

@ -15,8 +15,7 @@
# echo "" >> cloudflare_ips.lst
# curl -s https://www.cloudflare.com/ips-v6 >> cloudflare_ips.lst
#
# # Add to /etc/hosts (idempotent)
# grep -q "myapp.local" /etc/hosts || echo "127.0.0.1 myapp.local" | sudo tee -a /etc/hosts
# ```
#
# HOW TO START:
@ -47,7 +46,10 @@
services:
haproxy:
image: byjg/easy-haproxy:5.0.0
build:
context: ../..
dockerfile: build/Dockerfile
image: byjg/easy-haproxy:local
volumes:
- /var/run/docker.sock:/var/run/docker.sock
# Mount Cloudflare IP list
@ -62,11 +64,9 @@ services:
- "80:80/tcp"
- "1936:1936/tcp"
# Web application behind Cloudflare
# Web application behind Cloudflare (header-echo server for testing)
webapp:
image: byjg/static-httpserver
environment:
TITLE: "App Behind Cloudflare"
build: ./python-app
labels:
easyhaproxy.http.host: myapp.local
easyhaproxy.http.port: 80

14
examples/docker/docker-compose-ip-whitelist.yml
View file

@ -8,12 +8,7 @@
# - Custom HTTP status code for blocked requests
# - Admin panel or sensitive application protection
#
# REQUIREMENTS (run these first):
# ```bash
# # Add to /etc/hosts (idempotent)
# grep -q "admin.local" /etc/hosts || echo "127.0.0.1 admin.local" | sudo tee -a /etc/hosts
#
# # IMPORTANT: Update the allowed_ips in this file (line 52) with your actual IPs!
# # IMPORTANT: Update the easyhaproxy.http.plugin.ip_whitelist.allowed_ips with your actual IPs!
# # Default allows localhost and private networks for testing
# ```
#
@ -25,7 +20,7 @@
# HOW TO VERIFY IT'S WORKING:
# ```bash
# # Test from localhost (127.0.0.1 is whitelisted)
# curl http://admin.local/
# curl -k -H "Host: admin.local" http://127.0.0.1/
# # Expected: 200 OK - Access granted
#
# # Test from non-whitelisted IP
@ -47,7 +42,10 @@
services:
haproxy:
image: byjg/easy-haproxy:5.0.0
build:
context: ../../
dockerfile: build/Dockerfile
image: byjg/easy-haproxy:local
volumes:
- /var/run/docker.sock:/var/run/docker.sock
environment:

12
examples/docker/docker-compose-jwt-validator.yml
View file

@ -12,9 +12,6 @@
# ```bash
# # Generate SSL certificates and JWT keys (from project root)
# cd ../.. && ./examples/generate-keys.sh && cd examples/docker
#
# # Add to /etc/hosts (idempotent)
# grep -q "api.local" /etc/hosts || echo "127.0.0.1 api.local" | sudo tee -a /etc/hosts
# ```
#
# HOW TO START:
@ -25,7 +22,7 @@
# HOW TO VERIFY IT'S WORKING:
# ```bash
# # Test without token (should fail)
# curl http://api.local/
# curl -k -H "Host: api.local" http://127.0.0.1/
# # Expected: HTTP 403 - Missing Authorization HTTP header
#
# # Generate test JWT at https://jwt.io with:
@ -35,7 +32,7 @@
#
# # Test with valid token
# TOKEN="eyJhbGc..." # Replace with your generated token
# curl -H "Authorization: Bearer $TOKEN" http://api.local/
# curl -k -H "Host: host1.local" -H "Authorization: Bearer $TOKEN" http://api.local/
# # Expected: 200 OK with API response
#
# # View HAProxy stats
@ -53,7 +50,10 @@
services:
haproxy:
image: byjg/easy-haproxy:5.0.0
build:
context: ../..
dockerfile: build/Dockerfile
image: byjg/easy-haproxy:local
volumes:
- /var/run/docker.sock:/var/run/docker.sock
# Mount the public key for JWT verification

8
examples/docker/docker-compose-multi-containers.yml
View file

@ -46,7 +46,10 @@
services:
haproxy:
image: byjg/easy-haproxy:5.0.0
build:
context: ../../
dockerfile: build/Dockerfile
image: byjg/easy-haproxy:local
volumes:
- /var/run/docker.sock:/var/run/docker.sock
environment:
@ -56,7 +59,8 @@ services:
HAPROXY_PASSWORD: password
HAPROXY_STATS_PORT: 1936
ports:
- 19901:19901
- 19901:19901
- 1936:1936
nginx:

16
examples/docker/docker-compose-php-fpm.yml
View file

@ -9,11 +9,6 @@
# - PATH_INFO support for RESTful routing
# - Custom document root and index file configuration
#
# REQUIREMENTS (run these first):
# ```bash
# # Add to /etc/hosts (idempotent)
# grep -q "phpapp.local" /etc/hosts || echo "127.0.0.1 phpapp.local" | sudo tee -a /etc/hosts
# ```
#
# HOW TO START:
# ```bash
@ -23,15 +18,15 @@
# HOW TO VERIFY IT'S WORKING:
# ```bash
# # Test main page
# curl http://phpapp.local/
# curl -k -H "Host: phpapp.local" http://127.0.0.1/
# # Expected: 200 OK with PHP environment info
#
# # Test PHP info page
# curl http://phpapp.local/info.php
# -k -H "Host: phpapp.local" http://127.0.0.1/info.php
# # Expected: phpinfo() output
#
# # Test PATH_INFO routing
# curl http://phpapp.local/test-path-info.php/users/123
# -k -H "Host: phpapp.local" http://127.0.0.1/test-path-info.php/users/123
# # Expected: PATH_INFO=/users/123
#
# # View HAProxy stats
@ -49,7 +44,10 @@
services:
haproxy:
image: byjg/easy-haproxy:5.0.0
build:
context: ../../
dockerfile: build/Dockerfile
image: byjg/easy-haproxy:local
volumes:
- /var/run/docker.sock:/var/run/docker.sock
environment:

18
examples/docker/docker-compose-plugins-combined.yml
View file

@ -20,9 +20,6 @@
# echo "" >> cloudflare_ips.lst
# curl -s https://www.cloudflare.com/ips-v6 >> cloudflare_ips.lst
#
# # Add to /etc/hosts (idempotent)
# grep -q "website.local" /etc/hosts || echo "127.0.0.1 website.local api.local admin.local" | sudo tee -a /etc/hosts
# ```
#
# HOW TO START:
# ```bash
@ -32,21 +29,21 @@
# HOW TO VERIFY IT'S WORKING:
# ```bash
# # Test public website (Cloudflare + path blocking)
# curl http://website.local/
# curl -k -H "Host: website.local" http://127.0.0.1/
# # Expected: 200 OK
# curl http://website.local/admin
# curl -k -H "Host: website.local" http://127.0.0.1/admin
# # Expected: HTTP 404 - Path blocked
#
# # Test protected API (JWT required)
# curl http://api.local/
# curl -k -H "Host: api.local" http://127.0.0.1/
# # Expected: HTTP 403 - Missing Authorization header
# # Generate JWT at https://jwt.io (see jwt-validator example for details)
# TOKEN="eyJhbGc..." # Replace with your token
# curl -H "Authorization: Bearer $TOKEN" http://api.local/
# curl -H "Host: api.local" -H "Authorization: Bearer $TOKEN" http://127.0.0.1/
# # Expected: 200 OK
#
# # Test admin panel (IP whitelist)
# curl http://admin.local/
# curl -k -H "Host: admin.local" http://127.0.0.1/
# # Expected: 200 OK from localhost
#
# # View HAProxy stats
@ -65,7 +62,10 @@
services:
haproxy:
image: byjg/easy-haproxy:5.0.0
build:
context: ../../
dockerfile: build/Dockerfile
image: byjg/easy-haproxy:local
volumes:
- /var/run/docker.sock:/var/run/docker.sock
- ./cloudflare_ips.lst:/etc/haproxy/cloudflare_ips.lst:ro

8
examples/docker/docker-compose.yml
View file

@ -9,9 +9,6 @@
# - HAProxy stats interface
#
# REQUIREMENTS (run these first):
# ```bash
# # Add to /etc/hosts (idempotent)
# grep -q "host1.local" /etc/hosts || echo "127.0.0.1 host1.local host2.local" | sudo tee -a /etc/hosts
#
# # Generate SSL certificates
# cd ../.. && ./examples/generate-keys.sh && cd examples/docker
@ -51,7 +48,10 @@
services:
haproxy:
image: byjg/easy-haproxy:5.0.0
build:
context: ../..
dockerfile: build/Dockerfile
image: byjg/easy-haproxy:local
volumes:
- /var/run/docker.sock:/var/run/docker.sock
- ./host2.local.pem:/certs/haproxy/host2.local.pem

11
examples/docker/python-app/Dockerfile Normal file
View file

@ -0,0 +1,11 @@
FROM python:3.12-slim
WORKDIR /app
COPY server.py .
RUN chmod +x server.py
EXPOSE 8080
CMD ["python3", "server.py"]

33
examples/docker/python-app/server.py Normal file
View file

@ -0,0 +1,33 @@
#!/usr/bin/env python3
"""Simple HTTP server that echoes all request headers"""
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
class HeaderEchoHandler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
# Collect all headers
headers = dict(self.headers)
# Add the client IP as seen by this server
response = {
'headers': headers,
'client_ip': self.client_address[0],
'x_forwarded_for': self.headers.get('X-Forwarded-For', 'NOT SET')
}
self.wfile.write(json.dumps(response, indent=2).encode())
def log_message(self, format, *args):
# Log to stdout
print(f"{self.address_string()} - {format % args}")
if __name__ == '__main__':
port = 8080
server = HTTPServer(('0.0.0.0', port), HeaderEchoHandler)
print(f'Header echo server running on port {port}...')
server.serve_forever()

933
examples/docker/test_docker_compose.py Normal file
View file

@ -0,0 +1,933 @@
"""
Pytest test suite for EasyHAProxy Docker Compose examples
These tests verify the functionality of various docker-compose configurations.
Tests are organized by compose file and can be run individually or as a suite.
Requirements:
- pytest
- requests
- PyJWT
- cryptography
- docker-compose
Usage:
# Run all tests
pytest test_docker_compose.py -v
# Run specific test class
pytest test_docker_compose.py::TestBasicSSL -v
# Run specific test
pytest test_docker_compose.py::TestBasicSSL::test_https_host1 -v
# Run with markers
pytest test_docker_compose.py -m ssl -v
"""
import subprocess
import time
import os
from pathlib import Path
import pytest
import requests
import jwt as jwt_lib
from typing import Generator
# Base directory for docker-compose files
BASE_DIR = Path(__file__).parent.absolute()
@pytest.fixture(scope="session", autouse=True)
def generate_ssl_certificates():
"""
Generate SSL certificates once for all tests that require them.
This runs automatically at the start of the test session.
"""
script_path = BASE_DIR.parent / "generate-keys.sh"
# Check if script exists
if not script_path.exists():
pytest.skip(f"SSL certificate generation script not found: {script_path}")
# Run the script from the examples directory
result = subprocess.run(
["bash", str(script_path)],
cwd=BASE_DIR.parent,
capture_output=True,
text=True
)
if result.returncode != 0:
pytest.fail(f"Failed to generate SSL certificates:\n{result.stderr}")
yield
# No cleanup needed - certificates can be reused
class DockerComposeFixture:
"""Helper class to manage docker-compose lifecycle"""
def __init__(self, compose_file: str, startup_wait: int = 3, build: bool = True):
self.compose_file = str(BASE_DIR / compose_file)
self.startup_wait = startup_wait
self.build = build
def up(self):
"""Start docker-compose services"""
cmd = ["docker", "compose", "-f", self.compose_file, "up", "-d"]
if self.build:
cmd.append("--build")
subprocess.run(
cmd,
check=True,
capture_output=True
)
time.sleep(self.startup_wait)
def down(self):
"""Stop and remove docker-compose services"""
subprocess.run(
["docker", "compose", "-f", self.compose_file, "down", "--remove-orphans"],
check=True,
capture_output=True
)
@pytest.fixture
def docker_compose_basic_ssl() -> Generator[None, None, None]:
"""Fixture for docker-compose.yml (Basic SSL)"""
fixture = DockerComposeFixture("docker-compose.yml")
fixture.up()
yield
fixture.down()
@pytest.fixture
def docker_compose_jwt_validator() -> Generator[None, None, None]:
"""Fixture for docker-compose-jwt-validator.yml"""
fixture = DockerComposeFixture("docker-compose-jwt-validator.yml")
fixture.up()
yield
fixture.down()
@pytest.fixture
def docker_compose_multi_containers() -> Generator[None, None, None]:
"""Fixture for docker-compose-multi-containers.yml"""
fixture = DockerComposeFixture("docker-compose-multi-containers.yml")
fixture.up()
yield
fixture.down()
@pytest.fixture
def docker_compose_php_fpm() -> Generator[None, None, None]:
"""Fixture for docker-compose-php-fpm.yml"""
fixture = DockerComposeFixture("docker-compose-php-fpm.yml")
fixture.up()
yield
fixture.down()
@pytest.fixture
def docker_compose_plugins_combined() -> Generator[None, None, None]:
"""Fixture for docker-compose-plugins-combined.yml"""
fixture = DockerComposeFixture("docker-compose-plugins-combined.yml")
fixture.up()
yield
fixture.down()
@pytest.fixture
def docker_compose_ip_whitelist() -> Generator[None, None, None]:
"""Fixture for docker-compose-ip-whitelist.yml"""
fixture = DockerComposeFixture("docker-compose-ip-whitelist.yml")
fixture.up()
yield
fixture.down()
@pytest.fixture
def docker_compose_cloudflare() -> Generator[None, None, None]:
"""Fixture for docker-compose-cloudflare.yml"""
# Set up cloudflare_ips.lst with Docker network for testing
cloudflare_ips_path = BASE_DIR / "cloudflare_ips.lst"
# Download Cloudflare IPs
subprocess.run(
["curl", "-s", "https://www.cloudflare.com/ips-v4"],
stdout=open(cloudflare_ips_path, 'w'),
check=True
)
with open(cloudflare_ips_path, 'a') as f:
f.write("\n")
subprocess.run(
["curl", "-s", "https://www.cloudflare.com/ips-v6"],
stdout=open(cloudflare_ips_path, 'a'),
check=True
)
# Add Docker private network range so HAProxy treats test requests as from Cloudflare
# Docker bridge networks are typically in 172.16.0.0/12 range
with open(cloudflare_ips_path, 'a') as f:
f.write("\n")
f.write("172.16.0.0/12\n") # Docker private network range
fixture = DockerComposeFixture("docker-compose-cloudflare.yml")
fixture.up()
yield
fixture.down()
@pytest.fixture
def jwt_token() -> str:
"""Generate a valid JWT token for testing"""
private_key_path = BASE_DIR / "jwt_private.pem"
with open(private_key_path, 'r') as f:
private_key = f.read()
payload = {
'iss': 'https://auth.example.com/',
'aud': 'https://api.example.com',
'exp': 9999999999
}
token = jwt_lib.encode(payload, private_key, algorithm='RS256')
return token
# =============================================================================
# Test: docker-compose.yml - Basic SSL Setup
# =============================================================================
@pytest.mark.ssl
class TestBasicSSL:
"""Tests for basic SSL setup with two virtual hosts"""
def test_haproxy_config(self, docker_compose_basic_ssl):
"""Test HAProxy configuration has SSL and redirect configurations"""
result = subprocess.run(
["docker", "exec", "docker-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Test HTTPS backend for host1
https_host1_block = extract_backend_block(config, "srv_host1_local_443")
assert https_host1_block, "Backend srv_host1_local_443 not found"
assert "mode http" in https_host1_block
# Test HTTPS backend for host2
https_host2_block = extract_backend_block(config, "srv_host2_local_443")
assert https_host2_block, "Backend srv_host2_local_443 not found"
assert "mode http" in https_host2_block
# Verify SSL frontend exists and binds to port 443
assert "frontend https_in_443" in config or "bind *:443" in config
# Verify HTTP to HTTPS redirect
# Check for redirect rules in HTTP frontend or backends
assert "redirect scheme https" in config or "location: https://" in config
def test_https_host1(self, docker_compose_basic_ssl):
"""Test HTTPS access to host1.local"""
response = requests.get(
"https://127.0.0.1/",
headers={"Host": "host1.local"},
verify=False
)
assert response.status_code == 200
def test_https_host2(self, docker_compose_basic_ssl):
"""Test HTTPS access to host2.local"""
response = requests.get(
"https://127.0.0.1/",
headers={"Host": "host2.local"},
verify=False
)
assert response.status_code == 200
def test_http_redirect_host1(self, docker_compose_basic_ssl):
"""Test HTTP to HTTPS redirect for host1.local"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "host1.local"},
allow_redirects=False
)
assert response.status_code == 301
assert response.headers.get("location") == "https://host1.local/"
def test_http_redirect_host2(self, docker_compose_basic_ssl):
"""Test HTTP to HTTPS redirect for host2.local"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "host2.local"},
allow_redirects=False
)
assert response.status_code == 301
assert response.headers.get("location") == "https://host2.local/"
def test_haproxy_stats(self, docker_compose_basic_ssl):
"""Test HAProxy stats interface"""
response = requests.get(
"http://localhost:1936",
auth=("admin", "password")
)
assert response.status_code == 200
assert "Statistics Report for HAProxy" in response.text
# =============================================================================
# Test: docker-compose-jwt-validator.yml - JWT Validator Plugin
# =============================================================================
@pytest.mark.jwt
class TestJWTValidator:
"""Tests for JWT validator plugin"""
def test_haproxy_config(self, docker_compose_jwt_validator):
"""Test HAProxy configuration has JWT validator rules in the correct backend"""
result = subprocess.run(
["docker", "exec", "docker-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Extract the specific backend block
backend_block = extract_backend_block(config, "srv_api_local_80")
assert backend_block, "Backend srv_api_local_80 not found"
# Verify JWT validator plugin comment
assert "# JWT Validator - Validate JWT tokens" in backend_block
# Verify JWT validation rules
assert "http-request deny content-type 'text/html' string 'Missing Authorization HTTP header'" in backend_block
assert "http_auth_bearer,jwt_header_query('$.alg')" in backend_block
assert "http_auth_bearer,jwt_payload_query('$.iss')" in backend_block
assert "http_auth_bearer,jwt_payload_query('$.aud')" in backend_block
# Verify algorithm check
assert "var(txn.alg) -m str RS256" in backend_block
# Verify issuer and audience checks
assert "var(txn.iss) -m str https://auth.example.com/" in backend_block
assert "var(txn.aud) -m str https://api.example.com" in backend_block
# Verify JWT signature verification
assert 'jwt_verify(txn.alg,"/etc/haproxy/jwt_keys/api_pubkey.pem")' in backend_block
# Verify expiration check
assert "JWT has expired" in backend_block
def test_without_token(self, docker_compose_jwt_validator):
"""Test API access without JWT token (should fail)"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "api.local"}
)
assert response.status_code == 403
assert "Missing Authorization HTTP header" in response.text
def test_with_valid_token(self, docker_compose_jwt_validator, jwt_token):
"""Test API access with valid JWT token (should succeed)"""
response = requests.get(
"http://127.0.0.1/",
headers={
"Host": "api.local",
"Authorization": f"Bearer {jwt_token}"
}
)
assert response.status_code == 200
def test_haproxy_stats(self, docker_compose_jwt_validator):
"""Test HAProxy stats interface"""
response = requests.get(
"http://localhost:1936",
auth=("admin", "password")
)
assert response.status_code == 200
assert "Statistics Report for HAProxy" in response.text
# =============================================================================
# Test: docker-compose-multi-containers.yml - Load Balancing
# =============================================================================
@pytest.mark.loadbalancing
class TestMultiContainers:
"""Tests for load balancing with multiple container replicas"""
def test_haproxy_config(self, docker_compose_multi_containers):
"""Test HAProxy configuration has multiple backend servers for load balancing"""
result = subprocess.run(
["docker", "exec", "docker-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Extract the specific backend block
backend_block = extract_backend_block(config, "srv_www_helloworld_com_19901")
assert backend_block, "Backend srv_www_helloworld_com_19901 not found"
# Verify round-robin load balancing
assert "balance roundrobin" in backend_block
# Verify multiple servers are configured
server_lines = [line for line in backend_block.split('\n') if line.strip().startswith('server srv-')]
assert len(server_lines) >= 2, f"Expected at least 2 servers, found {len(server_lines)}"
# Verify both servers have check and weight
for server_line in server_lines:
assert "check" in server_line
assert "weight" in server_line
def test_load_balancing(self, docker_compose_multi_containers):
"""Test round-robin load balancing across replicas"""
container_ids = set()
for _ in range(6):
response = requests.get(
"http://localhost:19901/",
headers={"Host": "www.helloworld.com"}
)
assert response.status_code == 200
container_ids.add(response.text.strip())
# Should see at least 2 different container IDs
assert len(container_ids) >= 2
def test_domain_redirect(self, docker_compose_multi_containers):
"""Test domain redirect functionality"""
response = requests.get(
"http://localhost:19901/",
headers={"Host": "google.helloworld.com"},
allow_redirects=False
)
assert response.status_code == 301
assert response.headers.get("location") == "www.google.com/"
# =============================================================================
# Test: docker-compose-php-fpm.yml - PHP-FPM FastCGI Plugin
# =============================================================================
@pytest.mark.php
class TestPHPFPM:
"""Tests for PHP-FPM FastCGI plugin"""
def test_haproxy_config(self, docker_compose_php_fpm):
"""Test HAProxy configuration has FastCGI plugin configuration"""
result = subprocess.run(
["docker", "exec", "docker-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Extract the specific backend block
backend_block = extract_backend_block(config, "srv_phpapp_local_80")
assert backend_block, "Backend srv_phpapp_local_80 not found"
# Verify FastCGI app is used
assert "use-fcgi-app fcgi_phpapp_local" in backend_block
# Verify server uses fcgi protocol
assert "proto fcgi" in backend_block
# Verify port 9000 (PHP-FPM default)
assert ":9000" in backend_block
# Now check for fcgi-app configuration (not in backend, but in global config)
assert "fcgi-app fcgi_phpapp_local" in config
# Extract fcgi-app block
fcgi_lines = []
in_fcgi = False
for line in config.split('\n'):
if line.startswith('fcgi-app fcgi_phpapp_local'):
in_fcgi = True
elif in_fcgi:
if line.startswith(('fcgi-app ', 'frontend ', 'backend ', 'listen ')):
break
fcgi_lines.append(line)
fcgi_block = '\n'.join(fcgi_lines)
# Verify FastCGI plugin configuration
assert "docroot /var/www/html" in fcgi_block
assert "index index.php" in fcgi_block
assert "path-info" in fcgi_block
def test_main_page(self, docker_compose_php_fpm):
"""Test main PHP page"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "phpapp.local"}
)
assert response.status_code == 200
assert "PHP-FPM with EasyHAProxy" in response.text
def test_phpinfo(self, docker_compose_php_fpm):
"""Test PHP info page"""
response = requests.get(
"http://127.0.0.1/info.php",
headers={"Host": "phpapp.local"}
)
assert response.status_code == 200
assert "phpinfo()" in response.text
def test_path_info_routing(self, docker_compose_php_fpm):
"""Test PATH_INFO routing for RESTful URLs"""
response = requests.get(
"http://127.0.0.1/test-path-info.php/users/123",
headers={"Host": "phpapp.local"}
)
assert response.status_code == 200
assert "PATH_INFO" in response.text
assert "/users/123" in response.text
def test_haproxy_stats(self, docker_compose_php_fpm):
"""Test HAProxy stats interface"""
response = requests.get(
"http://localhost:1936",
auth=("admin", "password")
)
assert response.status_code == 200
assert "Statistics Report for HAProxy" in response.text
# =============================================================================
# Test: docker-compose-plugins-combined.yml - Multiple Plugins Combined
# =============================================================================
@pytest.mark.plugins
class TestPluginsCombined:
"""Tests for multiple plugins combined"""
def test_haproxy_config(self, docker_compose_plugins_combined):
"""Test HAProxy configuration has all plugin configurations in correct backends"""
result = subprocess.run(
["docker", "exec", "docker-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Test website backend (Cloudflare + deny_pages)
website_block = extract_backend_block(config, "srv_website_local_80")
assert website_block, "Backend srv_website_local_80 not found"
assert "# Cloudflare - Restore original visitor IP" in website_block
assert "acl from_cloudflare src -f /etc/haproxy/cloudflare_ips.lst" in website_block
assert "# Deny Pages - Block specific paths" in website_block
assert "acl denied_path path_beg /admin /wp-admin /wp-login.php /.env /config" in website_block
assert "http-request deny deny_status 404 if denied_path" in website_block
# Test API backend (JWT validator + deny_pages)
api_block = extract_backend_block(config, "srv_api_local_80")
assert api_block, "Backend srv_api_local_80 not found"
assert "# JWT Validator - Validate JWT tokens" in api_block
assert "Missing Authorization HTTP header" in api_block
assert "jwt_verify" in api_block
assert "# Deny Pages - Block specific paths" in api_block
assert "acl denied_path path_beg /internal /debug /metrics" in api_block
assert "http-request deny deny_status 403 if denied_path" in api_block
# Test admin backend (IP whitelist)
admin_block = extract_backend_block(config, "srv_admin_local_80")
assert admin_block, "Backend srv_admin_local_80 not found"
assert "# IP Whitelist - Only allow specific IPs" in admin_block
assert "acl whitelisted_ip src" in admin_block
assert "http-request deny deny_status 403 if !whitelisted_ip" in admin_block
def test_website_normal_access(self, docker_compose_plugins_combined):
"""Test normal access to public website"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "website.local"}
)
assert response.status_code == 200
def test_website_blocked_paths(self, docker_compose_plugins_combined):
"""Test blocked paths on public website"""
blocked_paths = ["/admin", "/wp-admin", "/.env", "/config"]
for path in blocked_paths:
response = requests.get(
f"http://127.0.0.1{path}",
headers={"Host": "website.local"}
)
assert response.status_code == 404
def test_api_without_token(self, docker_compose_plugins_combined):
"""Test API without JWT token"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "api.local"}
)
assert response.status_code == 403
assert "Missing Authorization HTTP header" in response.text
def test_api_with_valid_token(self, docker_compose_plugins_combined, jwt_token):
"""Test API with valid JWT token"""
response = requests.get(
"http://127.0.0.1/",
headers={
"Host": "api.local",
"Authorization": f"Bearer {jwt_token}"
}
)
assert response.status_code == 200
def test_api_blocked_paths_with_token(self, docker_compose_plugins_combined, jwt_token):
"""Test blocked paths on API even with valid JWT"""
blocked_paths = ["/internal", "/debug", "/metrics"]
for path in blocked_paths:
response = requests.get(
f"http://127.0.0.1{path}",
headers={
"Host": "api.local",
"Authorization": f"Bearer {jwt_token}"
}
)
assert response.status_code == 403
def test_admin_panel_localhost(self, docker_compose_plugins_combined):
"""Test admin panel from localhost (should be allowed)"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "admin.local"}
)
assert response.status_code == 200
def test_haproxy_stats(self, docker_compose_plugins_combined):
"""Test HAProxy stats interface"""
response = requests.get(
"http://localhost:1936",
auth=("admin", "password")
)
assert response.status_code == 200
assert "Statistics Report for HAProxy" in response.text
# =============================================================================
# Test: docker-compose-ip-whitelist.yml - IP Whitelist Plugin
# =============================================================================
def extract_backend_block(config: str, backend_name: str) -> str:
"""Extract a specific backend block from HAProxy configuration"""
lines = config.split('\n')
backend_lines = []
in_backend = False
for line in lines:
if line.startswith(f'backend {backend_name}'):
in_backend = True
backend_lines.append(line)
elif in_backend:
# Stop when we hit another backend, frontend, or global section
if line.startswith(('backend ', 'frontend ', 'global ', 'defaults ')):
break
backend_lines.append(line)
return '\n'.join(backend_lines)
@pytest.mark.security
class TestIPWhitelist:
"""Tests for IP whitelist plugin"""
def test_haproxy_config(self, docker_compose_ip_whitelist):
"""Test HAProxy configuration has IP whitelist rules in the correct backend"""
result = subprocess.run(
["docker", "exec", "docker-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Extract the specific backend block
backend_block = extract_backend_block(config, "srv_admin_local_80")
assert backend_block, "Backend srv_admin_local_80 not found"
# Verify IP whitelist plugin comment is in this backend
assert "# IP Whitelist - Only allow specific IPs" in backend_block
# Verify ACL for whitelisted IPs is in this backend
assert "acl whitelisted_ip src" in backend_block
# Extract the ACL line to verify IPs
acl_line = [line for line in backend_block.split('\n') if 'acl whitelisted_ip src' in line][0]
assert "127.0.0.1" in acl_line
assert "192.168.0.0/16" in acl_line
assert "10.0.0.0/8" in acl_line
assert "172.16.0.0/12" in acl_line
# Verify deny rule for non-whitelisted IPs is in this backend
assert "http-request deny deny_status 403 if !whitelisted_ip" in backend_block
def test_localhost_allowed(self, docker_compose_ip_whitelist):
"""Test access from localhost (should be allowed)"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "admin.local"}
)
assert response.status_code == 200
assert "Admin Panel" in response.text
def test_haproxy_stats(self, docker_compose_ip_whitelist):
"""Test HAProxy stats interface"""
response = requests.get(
"http://localhost:1936",
auth=("admin", "password")
)
assert response.status_code == 200
assert "Statistics Report for HAProxy" in response.text
# =============================================================================
# Test: docker-compose-cloudflare.yml - Cloudflare IP Restoration Plugin
# =============================================================================
@pytest.mark.cloudflare
class TestCloudflare:
"""Tests for Cloudflare IP restoration plugin"""
def test_haproxy_config(self, docker_compose_cloudflare):
"""Test HAProxy configuration has Cloudflare plugin rules in the correct backend"""
result = subprocess.run(
["docker", "exec", "docker-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Extract the specific backend block
backend_block = extract_backend_block(config, "srv_myapp_local_80")
assert backend_block, "Backend srv_myapp_local_80 not found"
# Verify Cloudflare plugin comment
assert "# Cloudflare - Restore original visitor IP" in backend_block
# Verify ACL for Cloudflare IPs
assert "acl from_cloudflare src -f /etc/haproxy/cloudflare_ips.lst" in backend_block
# Verify transaction variable for real IP
assert "http-request set-var(txn.real_ip) req.hdr(CF-Connecting-IP) if from_cloudflare" in backend_block
# Verify X-Forwarded-For header restoration with transaction variable
assert "http-request set-header X-Forwarded-For %[var(txn.real_ip)] if from_cloudflare" in backend_block
def test_normal_request(self, docker_compose_cloudflare):
"""
Test normal request without CF-Connecting-IP header
When a request comes from a Cloudflare IP (Docker network is in cloudflare_ips.lst)
but has NO CF-Connecting-IP header, the X-Forwarded-For will be empty because
HAProxy tries to extract from a non-existent header. This is expected behavior.
"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "myapp.local"}
)
assert response.status_code == 200
data = response.json()
assert 'headers' in data
assert 'x_forwarded_for' in data
# Verify X-Forwarded-For is empty (not a translated IP)
# Request comes from "Cloudflare IP" (Docker network) but has no CF-Connecting-IP
x_forwarded_for = data['x_forwarded_for']
assert x_forwarded_for == '', \
f"Expected X-Forwarded-For to be empty (no CF-Connecting-IP provided), got '{x_forwarded_for}'"
# Verify client_ip is the HAProxy container IP (backend sees connection from HAProxy)
client_ip = data['client_ip']
assert client_ip.startswith('172.'), \
f"Expected client_ip to be HAProxy container IP (172.x.x.x), got '{client_ip}'"
def test_cloudflare_ip_translation(self, docker_compose_cloudflare):
"""
Test that Cloudflare plugin actually translates CF-Connecting-IP to X-Forwarded-For
This test verifies the Cloudflare plugin correctly:
1. Detects requests from Cloudflare IPs (127.0.0.1 is in cloudflare_ips.lst)
2. Extracts the CF-Connecting-IP header value
3. Sets X-Forwarded-For header to that value
4. Backend receives the correct translated IP
"""
test_ip = "203.0.113.50"
response = requests.get(
"http://127.0.0.1/",
headers={
"Host": "myapp.local",
"CF-Connecting-IP": test_ip
}
)
assert response.status_code == 200
# Parse JSON response from header-echo server
data = response.json()
# VERIFY: X-Forwarded-For was set to the CF-Connecting-IP value
assert data['x_forwarded_for'] == test_ip, \
f"Expected X-Forwarded-For to be '{test_ip}', got '{data['x_forwarded_for']}'. " \
f"Cloudflare IP translation is NOT working!"
# Verify client_ip is still the HAProxy container IP (connection doesn't change)
client_ip = data['client_ip']
assert client_ip.startswith('172.'), \
f"Expected client_ip to be HAProxy container IP (172.x.x.x), got '{client_ip}'"
def test_haproxy_stats(self, docker_compose_cloudflare):
"""Test HAProxy stats interface"""
response = requests.get(
"http://localhost:1936",
auth=("admin", "password")
)
assert response.status_code == 200
assert "Statistics Report for HAProxy" in response.text
# =============================================================================
# Test: docker-compose-changed-label.yml - Custom Label Prefix
# =============================================================================
@pytest.fixture
def docker_compose_changed_label() -> Generator[None, None, None]:
"""Fixture for docker-compose-changed-label.yml"""
fixture = DockerComposeFixture("docker-compose-changed-label.yml")
fixture.up()
yield
fixture.down()
@pytest.mark.custom_label
class TestChangedLabel:
"""Tests for docker-compose-changed-label.yml - Custom label prefix"""
def test_haproxy_config(self, docker_compose_changed_label):
"""Test HAProxy configuration with custom label prefix"""
result = subprocess.run(
["docker", "exec", "docker-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Verify HTTPS backend exists
assert "backend srv_host1_local_443" in config
# Verify SSL configuration (frontend with SSL)
assert "bind *:443" in config
assert "ssl crt" in config
# Verify HTTP backend exists
assert "backend srv_host1_local_80" in config
# Verify HTTP to HTTPS redirect is configured
assert "redirect prefix https://host1.local code 301" in config
def test_https_access(self, docker_compose_changed_label):
"""Test HTTPS access to host1.local"""
response = requests.get(
"https://127.0.0.1/",
headers={"Host": "host1.local"},
verify=False # Self-signed certificate
)
assert response.status_code == 200
# byjg/static-httpserver returns a "Coming Soon" page
assert "soon" in response.text.lower() or "coming" in response.text.lower()
def test_http_redirect(self, docker_compose_changed_label):
"""Test HTTP to HTTPS redirect"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "host1.local"},
allow_redirects=False
)
# The redirect uses 301 (permanent) as configured in the labels
assert response.status_code == 301
assert response.headers["Location"] == "https://host1.local/"
def test_custom_label_prefix(self, docker_compose_changed_label):
"""Verify custom label prefix 'haproxy' is being used"""
# Get container ID for static-httpserver
result = subprocess.run(
["docker", "ps", "-q", "-f", "ancestor=byjg/static-httpserver"],
capture_output=True,
text=True,
check=True
)
container_id = result.stdout.strip()
assert container_id, "Container not found"
# Inspect container labels
result = subprocess.run(
["docker", "inspect", container_id],
capture_output=True,
text=True,
check=True
)
# Verify labels start with "haproxy." not "easyhaproxy."
assert '"haproxy.http.host":' in result.stdout or '"haproxy.http.host"' in result.stdout
assert '"haproxy.https.host":' in result.stdout or '"haproxy.https.host"' in result.stdout
def test_haproxy_stats(self, docker_compose_changed_label):
"""Test HAProxy stats interface"""
response = requests.get(
"http://localhost:1936",
auth=("admin", "password")
)
assert response.status_code == 200
assert "Statistics Report for HAProxy" in response.text
# =============================================================================
# Helper functions for manual testing
# =============================================================================
def run_manual_test(compose_file: str, test_function):
"""
Helper function to run a test manually without pytest
Example:
def my_test():
response = requests.get("http://localhost/")
assert response.status_code == 200
run_manual_test("docker-compose.yml", my_test)
"""
fixture = DockerComposeFixture(compose_file)
try:
fixture.up()
test_function()
print("✅ Test passed!")
except AssertionError as e:
print(f"❌ Test failed: {e}")
finally:
fixture.down()
if __name__ == "__main__":
print("This is a pytest test suite. Run with: pytest test_docker_compose.py -v")
print("\nAvailable test classes:")
print(" - TestBasicSSL: Basic SSL setup tests")
print(" - TestJWTValidator: JWT validator plugin tests")
print(" - TestMultiContainers: Load balancing tests")
print(" - TestPHPFPM: PHP-FPM FastCGI tests")
print(" - TestPluginsCombined: Combined plugins tests")
print(" - TestIPWhitelist: IP whitelist plugin tests")
print(" - TestCloudflare: Cloudflare IP restoration plugin tests")
print(" - TestChangedLabel: Custom label prefix tests")