tests/test_analyzer.py

"""Tests for the Analyzer module."""

import pytest

from intelliscan.modules.analyzer import Analyzer


def test_sqli_detection_dump(sample_injection_results):
    analyzer = Analyzer([sample_injection_results[0]])
    results = analyzer.run()
    assert len(results) == 1
    r = results[0]
    assert r.label == "VULNERABLE"
    assert r.severity == "HIGH"
    assert "Data dump" in r.reason


def test_sqli_no_vulnerability(sample_injection_results):
    analyzer = Analyzer([sample_injection_results[1]])
    results = analyzer.run()
    assert results[0].label == "NOT_VULNERABLE"


def test_baseline_suppresses_form_label_false_positive():
    """Dump signatures present in the benign baseline (e.g. the 'Username:' /
    'Password:' form labels) must not be counted as an injected data dump."""
    page = (
        "<form><label>Username:</label><input name='username'>"
        "<label>Username:</label><label>Password:</label></form>"
        "Username and/or password incorrect."
    )
    entry = {
        "target_url": "http://localhost:8080/vulnerabilities/brute/",
        "method": "GET",
        "vuln_type": "sqli",
        "param": "username",
        "payload": "' OR '1'='1",
        "status_code": 200,
        "response_len": len(page),
        "response_body": page,  # identical to baseline — no extra rows leaked
    }
    baselines = {"http://localhost:8080/vulnerabilities/brute/": page}
    results = Analyzer([entry], baselines=baselines).run()
    assert results[0].label == "NOT_VULNERABLE"


def test_baseline_still_flags_real_dump():
    """Rows leaked beyond the baseline are still detected as a dump."""
    baseline = "<label>First name:</label>"  # one label in chrome
    dumped = baseline + "First name: admin First name: gordon First name: pablo"
    entry = {
        "target_url": "http://localhost:8080/vulnerabilities/sqli/",
        "method": "GET",
        "vuln_type": "sqli",
        "param": "id",
        "payload": "' OR '1'='1",
        "status_code": 200,
        "response_len": len(dumped),
        "response_body": dumped,
    }
    baselines = {"http://localhost:8080/vulnerabilities/sqli/": baseline}
    results = Analyzer([entry], baselines=baselines).run()
    assert results[0].label == "VULNERABLE"
    assert "Data dump" in results[0].reason


def test_reflected_union_payload_not_a_dump():
    """A page that reflects a UNION payload verbatim (e.g. the stored-XSS
    guestbook) must not be flagged as a SQLi data dump just because the payload
    text itself contains 'information_schema'."""
    payload = "' UNION SELECT null,table_name FROM information_schema.tables-- -"
    # guestbook echoes the stored message twice (message list + retained input)
    page = f"<div>Message: {payload}</div><textarea>{payload}</textarea>"
    entry = {
        "target_url": "http://localhost:8080/vulnerabilities/xss_s/",
        "method": "POST",
        "vuln_type": "sqli",
        "param": "mtxMessage",
        "payload": payload,
        "status_code": 200,
        "response_len": len(page),
        "response_body": page,
    }
    baselines = {"http://localhost:8080/vulnerabilities/xss_s/": ""}
    results = Analyzer([entry], baselines=baselines).run()
    assert results[0].label == "NOT_VULNERABLE"


def test_sqli_mysql_error_detection():
    """MySQL error message triggers VULNERABLE classification."""
    entry = {
        "target_url": "http://example.com/page",
        "method": "GET",
        "vuln_type": "sqli",
        "param": "id",
        "payload": "' OR '1'='1",
        "status_code": 500,
        "response_len": 900,
        "response_body": "You have an error in your SQL syntax near ''1'='1'",
    }
    results = Analyzer([entry]).run()
    assert results[0].label == "VULNERABLE"
    assert "sql error" in results[0].reason.lower()


def test_sqli_mssql_error_detection():
    """MSSQL 'Incorrect syntax near' triggers VULNERABLE classification."""
    entry = {
        "target_url": "http://example.com/page",
        "method": "GET",
        "vuln_type": "sqli",
        "param": "q",
        "payload": "' OR 1=1--",
        "status_code": 500,
        "response_len": 600,
        "response_body": "Incorrect syntax near 'OR'. Unclosed quotation mark.",
    }
    results = Analyzer([entry]).run()
    assert results[0].label == "VULNERABLE"


def test_sqli_postgres_error_detection():
    """PostgreSQL syntax error triggers VULNERABLE classification."""
    entry = {
        "target_url": "http://example.com/search",
        "method": "GET",
        "vuln_type": "sqli",
        "param": "q",
        "payload": "' UNION SELECT 1--",
        "status_code": 500,
        "response_len": 700,
        "response_body": 'pg_query(): Query failed: ERROR: syntax error at or near "UNION"',
    }
    results = Analyzer([entry]).run()
    assert results[0].label == "VULNERABLE"


def test_sqli_information_schema_leak():
    """INFORMATION_SCHEMA in response body indicates data exfiltration."""
    entry = {
        "target_url": "http://example.com/page",
        "method": "GET",
        "vuln_type": "sqli",
        "param": "id",
        "payload": "1 UNION SELECT table_name,2 FROM information_schema.tables--",
        "status_code": 200,
        "response_len": 2000,
        "response_body": (
            "information_schema information_schema information_schema " "users admins sessions"
        ),
    }
    results = Analyzer([entry]).run()
    assert results[0].label == "VULNERABLE"


def test_xss_reflection(sample_injection_results):
    analyzer = Analyzer([sample_injection_results[2]])
    results = analyzer.run()
    assert results[0].label == "VULNERABLE"
    assert "reflected" in results[0].reason.lower()


def test_xss_partial_pattern_reflection():
    """Dangerous XSS token reflected even when full payload is sanitised."""
    entry = {
        "target_url": "http://example.com/greet",
        "method": "GET",
        "vuln_type": "xss_r",
        "param": "name",
        "payload": "<script>alert(1)</script>",
        "status_code": 200,
        "response_len": 500,
        "response_body": "Hello &lt;script>alert(1)</script>, welcome!",
    }
    # body contains <script (from partial decode), payload contains <script
    results = Analyzer([entry]).run()
    assert results[0].label == "VULNERABLE"


def test_lfi_passwd_signature(sample_injection_results):
    analyzer = Analyzer([sample_injection_results[3]])
    results = analyzer.run()
    assert results[0].label == "VULNERABLE"
    assert "root:x:0:0" in results[0].reason


def test_lfi_windows_system32():
    """Windows system path leak triggers LFI classification."""
    entry = {
        "target_url": "http://example.com/download",
        "method": "GET",
        "vuln_type": "lfi",
        "param": "file",
        "payload": "C:/windows/system32/drivers/etc/hosts",
        "status_code": 200,
        "response_len": 1200,
        "response_body": "Contents of c:\\windows\\system32\\drivers\\etc\\hosts:\n127.0.0.1 localhost",
    }
    results = Analyzer([entry]).run()
    assert results[0].label == "VULNERABLE"


def test_lfi_blocked_path_traversal(sample_injection_results):
    analyzer = Analyzer([sample_injection_results[4]])
    results = analyzer.run()
    assert results[0].label == "NOT_VULNERABLE"


def test_unknown_vuln_type_returns_not_vulnerable():
    """Unsupported vuln_type should default to NOT_VULNERABLE."""
    entry = {
        "target_url": "http://example.com/",
        "method": "GET",
        "vuln_type": "csrf",
        "param": "token",
        "payload": "x",
        "status_code": 200,
        "response_len": 100,
        "response_body": "OK",
    }
    results = Analyzer([entry]).run()
    assert results[0].label == "NOT_VULNERABLE"


def test_stats_by_type(sample_injection_results):
    analyzer = Analyzer(sample_injection_results)
    analyzer.run()
    stats = analyzer.stats_by_type()
    assert stats["sqli"]["total"] == 2
    assert stats["sqli"]["vulnerable"] == 1
    assert stats["lfi"]["total"] == 2
    assert stats["lfi"]["vulnerable"] == 1


def test_labeled_result_has_required_fields(sample_injection_results):
    """Every LabeledResult must carry all schema fields."""
    results = Analyzer(sample_injection_results).run()
    for r in results:
        assert r.label in {"VULNERABLE", "NOT_VULNERABLE"}
        assert r.severity in {"HIGH", "MEDIUM", "LOW", "INFO"}
        assert isinstance(r.reason, str) and r.reason
        assert isinstance(r.response_len, int)