Spaces:

techprotrade
/

annator-command-center

Running

File size: 19,678 Bytes

aef804e

"""
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
        "<script>alert('XSS')</script>",
        "<img src=x onerror=alert('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()