OWASP Top 10 Test Cases Every Developer Should Write

import pytest
import requests

BASE_URL = "http://localhost:8080"

def test_horizontal_privilege_escalation():
    # Log in as user A, get their auth token
    user_a = requests.post(f"{BASE_URL}/auth/login", json={
        "email": "user_a@example.com",
        "password": "password123"
    })
    token_a = user_a.json()["token"]

    # Log in as user B to get their resource ID
    user_b = requests.post(f"{BASE_URL}/auth/login", json={
        "email": "user_b@example.com",
        "password": "password456"
    })
    token_b = user_b.json()["token"]
    user_b_profile_id = user_b.json()["user"]["id"]

    # User A attempts to access User B's profile data
    response = requests.get(
        f"{BASE_URL}/api/users/{user_b_profile_id}/private-data",
        headers={"Authorization": f"Bearer {token_a}"}
    )

    assert response.status_code == 403, (
        f"Expected 403 Forbidden, got {response.status_code}. "
        "User A should not access User B's private data."
    )

def test_admin_endpoint_blocked_for_regular_user():
    user = requests.post(f"{BASE_URL}/auth/login", json={
        "email": "regular_user@example.com",
        "password": "password123"
    })
    token = user.json()["token"]

    response = requests.get(
        f"{BASE_URL}/admin/users",
        headers={"Authorization": f"Bearer {token}"}
    )

    assert response.status_code in [401, 403]

import ssl
import socket
import hashlib

def test_tls_enforced_no_http_fallback():
    """Verify the application does not serve content over plain HTTP."""
    import urllib.request
    try:
        response = urllib.request.urlopen("http://yourapp.example.com/api/health")
        # If we get here, HTTP is serving content — that's a failure
        assert False, "Application should not serve content over plain HTTP"
    except Exception as e:
        # Expect a redirect to HTTPS or a connection refusal
        assert "301" in str(e) or "302" in str(e) or "Connection refused" in str(e)

def test_password_not_returned_in_api_response():
    response = requests.get(
        f"{BASE_URL}/api/users/me",
        headers={"Authorization": f"Bearer {valid_token}"}
    )
    data = response.json()
    assert "password" not in data
    assert "password_hash" not in data
    assert "pwd" not in data

def test_weak_tls_versions_rejected():
    """Verify TLS 1.0 and 1.1 are not accepted."""
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    context.maximum_version = ssl.TLSVersion.TLSv1_1
    context.check_hostname = False
    context.verify_mode = ssl.CERT_NONE

    try:
        with socket.create_connection(("yourapp.example.com", 443)) as sock:
            with context.wrap_socket(sock) as ssock:
                assert False, "TLS 1.1 should be rejected"
    except ssl.SSLError:
        pass  # Expected — server correctly rejected weak TLS

def test_sql_injection_in_search_parameter():
    """Verify search endpoint is not vulnerable to SQL injection."""
    payloads = [
        "' OR '1'='1",
        "'; DROP TABLE users; --",
        "' UNION SELECT username, password FROM users --",
        "1' AND SLEEP(5) --",
    ]

    for payload in payloads:
        response = requests.get(
            f"{BASE_URL}/api/products/search",
            params={"q": payload},
            headers={"Authorization": f"Bearer {valid_token}"}
        )
        # Should return empty results or 400, never 500 or user data
        assert response.status_code != 500, (
            f"Payload '{payload}' caused a 500 error — likely an unhandled injection"
        )
        if response.status_code == 200:
            data = response.json()
            # Should not return more results than expected for a nonsense query
            assert len(data.get("results", [])) == 0 or all(
                "password" not in str(item) for item in data["results"]
            )

def test_login_rate_limiting():
    """Brute force protection should kick in after repeated failed logins."""
    responses = []
    for i in range(20):
        r = requests.post(f"{BASE_URL}/auth/login", json={
            "email": "victim@example.com",
            "password": f"wrong_password_{i}"
        })
        responses.append(r.status_code)

    # After enough failures, should see 429 Too Many Requests
    assert 429 in responses, (
        "No rate limiting detected on login endpoint after 20 failed attempts"
    )

def test_password_reset_token_single_use():
    """Password reset tokens should be invalidated after first use."""
    # Trigger password reset
    requests.post(f"{BASE_URL}/auth/forgot-password", json={"email": "user@example.com"})
    token = get_reset_token_from_email()  # Helper to read test inbox

    # Use the token once
    requests.post(f"{BASE_URL}/auth/reset-password", json={
        "token": token,
        "new_password": "NewSecurePassword1!"
    })

    # Attempt to reuse the same token
    response = requests.post(f"{BASE_URL}/auth/reset-password", json={
        "token": token,
        "new_password": "AnotherPassword2@"
    })
    assert response.status_code in [400, 401, 410], "Reset token should be invalidated after use"

def test_security_headers_present():
    response = requests.get(f"{BASE_URL}/")
    headers = response.headers

    required_headers = {
        "X-Content-Type-Options": "nosniff",
        "X-Frame-Options": lambda v: v in ["DENY", "SAMEORIGIN"],
        "Strict-Transport-Security": lambda v: "max-age=" in v,
        "Content-Security-Policy": lambda v: len(v) > 0,
    }

    for header, expected in required_headers.items():
        assert header in headers, f"Missing security header: {header}"
        if callable(expected):
            assert expected(headers[header]), (
                f"Header {header} has unexpected value: {headers[header]}"
            )
        else:
            assert headers[header] == expected

def test_server_header_not_verbose():
    response = requests.get(f"{BASE_URL}/")
    server_header = response.headers.get("Server", "")
    # Should not expose exact version numbers
    assert not any(char.isdigit() for char in server_header) or server_header == "", (
        f"Server header exposes version info: {server_header}"
    )

def test_directory_listing_disabled():
    response = requests.get(f"{BASE_URL}/static/")
    assert response.status_code != 200 or "Index of" not in response.text

# In your CI pipeline (GitHub Actions example)
- name: Audit npm dependencies
  run: npm audit --audit-level=high

- name: Check Python dependencies
  run: |
    pip install safety
    safety check --full-report

- name: OWASP Dependency Check
  uses: dependency-check/Dependency-Check_Action@main
  with:
    project: 'my-app'
    path: '.'
    format: 'HTML'
    args: '--failOnCVSS 7'

def test_session_token_entropy():
    """Session tokens should be sufficiently random and unpredictable."""
    import re
    tokens = []
    for _ in range(5):
        r = requests.post(f"{BASE_URL}/auth/login", json={
            "email": "user@example.com",
            "password": "password123"
        })
        token = r.json()["token"]
        tokens.append(token)

    # All tokens should be unique
    assert len(set(tokens)) == len(tokens), "Session tokens are not unique"

    # Tokens should meet minimum length (JWT or equivalent)
    for token in tokens:
        assert len(token) >= 32, f"Token too short: {len(token)} chars"

def test_session_invalidated_on_logout():
    login_response = requests.post(f"{BASE_URL}/auth/login", json={
        "email": "user@example.com",
        "password": "password123"
    })
    token = login_response.json()["token"]

    # Verify token works
    assert requests.get(
        f"{BASE_URL}/api/me",
        headers={"Authorization": f"Bearer {token}"}
    ).status_code == 200

    # Log out
    requests.post(f"{BASE_URL}/auth/logout", headers={"Authorization": f"Bearer {token}"})

    # Token should no longer be valid
    response = requests.get(
        f"{BASE_URL}/api/me",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 401, "Token still valid after logout"

def test_deserialization_of_user_input_rejected():
    """Endpoints should not deserialize arbitrary user-controlled objects."""
    # Python pickle payload example — application should reject or ignore
    import pickle, base64

    class Exploit:
        def __reduce__(self):
            return (eval, ("__import__('os').system('id')",))

    payload = base64.b64encode(pickle.dumps(Exploit())).decode()

    response = requests.post(
        f"{BASE_URL}/api/import",
        json={"data": payload, "format": "pickle"}
    )
    # Should reject with 400 or 415, not execute the payload
    assert response.status_code in [400, 415, 422]

def test_failed_login_generates_audit_log():
    """Failed authentication attempts should be logged."""
    # Trigger a failed login
    requests.post(f"{BASE_URL}/auth/login", json={
        "email": "user@example.com",
        "password": "wrong_password"
    })

    # Check the audit log (requires access to log storage in test environment)
    logs = get_recent_audit_logs()  # Your log query helper
    failed_login_events = [
        log for log in logs
        if log.get("event_type") == "auth.failure"
        and log.get("email") == "user@example.com"
    ]
    assert len(failed_login_events) >= 1, "Failed login not recorded in audit log"

def test_ssrf_protection_on_url_fetch_endpoint():
    """URL fetch endpoints should block internal/private IP ranges."""
    internal_urls = [
        "http://169.254.169.254/latest/meta-data/",  # AWS metadata
        "http://localhost/admin",
        "http://127.0.0.1:6379",  # Redis
        "http://10.0.0.1/internal-api",
        "http://192.168.1.1/router-admin",
    ]

    for url in internal_urls:
        response = requests.post(
            f"{BASE_URL}/api/fetch-preview",
            json={"url": url},
            headers={"Authorization": f"Bearer {valid_token}"}
        )
        assert response.status_code in [400, 403, 422], (
            f"Internal URL '{url}' was not blocked. Got {response.status_code}"
        )

name: Security Tests
on: [pull_request]

jobs:
  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Start application
        run: docker-compose up -d
      - name: Wait for app
        run: sleep 10
      - name: Run OWASP security tests
        run: pytest tests/security/ -v --tb=short
      - name: Upload results
        if: always()
        uses: actions/upload-artifact@v3
        with:
          name: security-test-results
          path: test-results/