rules.py

import datetime
import time

# API definitions and risk scoring (initial scoring assisted by AI, manually adjusted)

# Memory manipulation APIs – used for allocation, protection changes and shellcode handling
memory_apis = {
    "VirtualAlloc": 30,
    "VirtualAllocEx": 35,
    "VirtualProtect": 25,
    "VirtualProtectEx": 25,
    "WriteProcessMemory": 45,
    "ReadProcessMemory": 20,
    "NtAllocateVirtualMemory": 35,
    "NtWriteVirtualMemory": 45
}

# Process injection APIs – indicate cross-process interaction or code execution
injection_apis = {
    "CreateRemoteThread": 50,
    "CreateRemoteThreadEx": 50,
    "NtCreateThreadEx": 50,
    "OpenProcess": 30,
    "NtOpenProcess": 30,
    "QueueUserAPC": 45,
    "SetThreadContext": 45,
    "GetThreadContext": 20,
    "SuspendThread": 25,
    "ResumeThread": 25
}

# Execution APIs – process spawning and command execution
execution_apis = {
    "CreateProcessA": 20,
    "CreateProcessW": 20,
    "WinExec": 25,
    "ShellExecuteA": 15,
    "ShellExecuteW": 15,
    "system": 25
}

# Dynamic DLL loading – often used in loaders or obfuscation
dll_apis = {
    "LoadLibraryA": 20,
    "LoadLibraryW": 20,
    "LoadLibraryExA": 25,
    "LoadLibraryExW": 25,
    "GetProcAddress": 25,
    "LdrLoadDll": 25
}

# Persistence via registry
persistence_apis = {
    "RegOpenKeyExA": 10,
    "RegOpenKeyExW": 10,
    "RegSetValueExA": 20,
    "RegSetValueExW": 20,
    "RegCreateKeyExA": 20,
    "RegCreateKeyExW": 20
}

# File system interaction – low risk alone, higher in context
file_apis = {
    "CreateFileA": 5,
    "CreateFileW": 5,
    "WriteFile": 5,
    "ReadFile": 5,
    "DeleteFileA": 10,
    "DeleteFileW": 10
}

# Match imported APIs against a scoring dictionary
def match_apis(imports, api_dict):
    return {api: api_dict[api] for api in api_dict if api in imports}


# Main evaluation logic – combines API scoring and behavioral heuristics
def evaluate_rules(file_info):
    imports = file_info["imports"]

    # signature metadata (trust model)
    signature = file_info.get("signature", "unsigned")   # unsigned | selfsigned | valid
    publisher = file_info.get("publisher", None)

    total_score = 0
    findings = []

    # trusted vendors list (highest trust level)
    trusted_publishers = [
        "Microsoft Corporation",
        "Google LLC",
        "Adobe Inc."
    ]

    # trusted signed binary overrides all scoring
    if signature == "valid" and publisher in trusted_publishers:
        return 0, [{
            "type": "Trusted Vendor Signature",
            "score": 0,
            "matches": {
                "publisher": publisher
            }
        }]

    # valid signature but unknown publisher (low trust)
    if signature == "valid" and publisher not in trusted_publishers:
        total_score += 20
        findings.append({
            "type": "Untrusted Signed Binary",
            "score": 20,
            "matches": {
                "publisher": publisher
            }
        })

    # self-signed binary (common in malware or internal tools)
    elif signature == "selfsigned":
        total_score += 40
        findings.append({
            "type": "Self-Signed Binary",
            "score": 40,
            "matches": {}
        })

    # unsigned binary (no trust signals)
    elif signature == "unsigned":
        total_score += 40
        findings.append({
            "type": "Unsigned Binary",
            "score": 40,
            "matches": {}
        })

    # API category scoring (behavioral signals)
    api_groups = [
        ("Memory APIs", memory_apis),
        ("Injection APIs", injection_apis),
        ("Execution APIs", execution_apis),
        ("DLL APIs", dll_apis),
        ("Persistence APIs", persistence_apis),
        ("File APIs", file_apis),
    ]

    for name, api_dict in api_groups:
        matches = match_apis(imports, api_dict)

        if matches:
            score = sum(matches.values())
            total_score += score

            findings.append({
                "type": name,
                "score": score,
                "matches": matches
            })

    # classic process injection pattern detection
    if (
        "VirtualAlloc" in imports and
        "WriteProcessMemory" in imports and
        "CreateRemoteThread" in imports
    ):
        total_score += 80
        findings.append({
            "type": "Injection Chain",
            "score": 80,
            "matches": {
                "VirtualAlloc": 30,
                "WriteProcessMemory": 45,
                "CreateRemoteThread": 50
            }
        })

    # stealth injection using NT APIs
    if (
        "OpenProcess" in imports and
        "NtCreateThreadEx" in imports
    ):
        total_score += 60
        findings.append({
            "type": "Stealth Injection",
            "score": 60,
            "matches": {
                "OpenProcess": 30,
                "NtCreateThreadEx": 50
            }
        })

    # dynamic API resolution typical for loaders
    if (
        ("LoadLibraryA" in imports or "LoadLibraryW" in imports) and
        "GetProcAddress" in imports
    ):
        total_score += 40
        findings.append({
            "type": "Dynamic API Resolution",
            "score": 40,
            "matches": {
                "LoadLibrary": 20,
                "GetProcAddress": 25
            }
        })

    # PE section and overlay/packer heuristics
    section_analysis = file_info.get("section_analysis", {})
    overlay = section_analysis.get("overlay", {})
    suspicious_section_names = section_analysis.get("suspicious_section_names", [])
    invalid_sections = section_analysis.get("invalid_sections", [])
    packer_section_names = section_analysis.get("packer_section_names", [])
    packer_signatures = section_analysis.get("packer_signatures", [])
    known_sections = section_analysis.get("known_sections", {})

    if overlay.get("present"):
        overlay_score = 50 if overlay.get("size", 0) >= 8192 else 30
        total_score += overlay_score
        findings.append({
            "type": "Overlay Detected",
            "score": overlay_score,
            "matches": {
                "overlay_size": overlay.get("size"),
                "overlay_entropy": overlay.get("entropy")
            }
        })

    if invalid_sections:
        total_score += 40
        findings.append({
            "type": "Invalid Section Names",
            "score": 40,
            "matches": {
                "sections": ", ".join(invalid_sections)
            }
        })

    if suspicious_section_names:
        total_score += 25
        findings.append({
            "type": "Suspicious Section Names",
            "score": 25,
            "matches": {
                "sections": ", ".join(suspicious_section_names)
            }
        })

    if packer_section_names:
        total_score += 50
        findings.append({
            "type": "Packer Section Names",
            "score": 50,
            "matches": {
                "sections": ", ".join(packer_section_names)
            }
        })

    if packer_signatures:
        total_score += 60
        findings.append({
            "type": "Packer / Obfuscator Signature",
            "score": 60,
            "matches": {
                "signatures": ", ".join(packer_signatures)
            }
        })

    if known_sections:
        for section_label in [".text", ".rdata", ".rsrc"]:
            section_info = known_sections.get(section_label)
            if section_info and section_info.get("entropy") is not None and section_info["entropy"] >= 7.0:
                total_score += 15
                findings.append({
                    "type": f"High Entropy {section_label} Section",
                    "score": 15,
                    "matches": {
                        "entropy": section_info["entropy"]
                    }
                })

    # Entropy scoring: high entropy often indicates packing or obfuscation
    entropy_info = file_info.get("entropy", {})
    file_entropy = entropy_info.get("file_entropy")
    max_section_entropy = entropy_info.get("max_section_entropy")
    high_entropy_sections = entropy_info.get("high_entropy_sections", [])

    if file_entropy is not None and file_entropy >= 7.0:
        bonus = 30 if file_entropy >= 7.5 else 20
        total_score += bonus
        findings.append({
            "type": "High File Entropy",
            "score": bonus,
            "matches": {
                "file_entropy": file_entropy
            }
        })

    if max_section_entropy is not None and max_section_entropy >= 7.0:
        total_score += 40
        findings.append({
            "type": "High Section Entropy",
            "score": 40,
            "matches": {
                "max_section_entropy": max_section_entropy,
                "sections": ", ".join([s["name"] for s in high_entropy_sections[:3]])
            }
        })

    # Behavioral scoring: correlated API patterns that increase suspicion
    persistence_matches = match_apis(imports, persistence_apis)
    injection_matches = match_apis(imports, injection_apis)
    execution_matches = match_apis(imports, execution_apis)
    file_matches = match_apis(imports, file_apis)
    memory_matches = match_apis(imports, memory_apis)

    if persistence_matches and injection_matches:
        total_score += 60
        findings.append({
            "type": "Persistence + Injection",
            "score": 60,
            "matches": {
                **persistence_matches,
                **injection_matches
            }
        })

    if (
        ("CreateProcessA" in imports or "CreateProcessW" in imports) and
        ("WriteProcessMemory" in imports or "SetThreadContext" in imports or "ResumeThread" in imports or "NtCreateThreadEx" in imports)
    ):
        total_score += 80
        findings.append({
            "type": "Process Hollowing / Remote Injection",
            "score": 80,
            "matches": {
                **match_apis(imports, {"CreateProcessA": 20, "CreateProcessW": 20, "WriteProcessMemory": 45, "SetThreadContext": 45, "ResumeThread": 25, "NtCreateThreadEx": 50})
            }
        })

    if persistence_matches and execution_matches:
        total_score += 40
        findings.append({
            "type": "Persistence + Execution",
            "score": 40,
            "matches": {
                **persistence_matches,
                **execution_matches
            }
        })

    if file_matches and persistence_matches:
        total_score += 30
        findings.append({
            "type": "Dropper / Installer Behavior",
            "score": 30,
            "matches": {
                **file_matches,
                **persistence_matches
            }
        })

    if (("GetProcAddress" in imports or "LoadLibraryA" in imports or "LoadLibraryW" in imports or "LdrLoadDll" in imports) and "VirtualProtect" in imports):
        total_score += 45
        findings.append({
            "type": "Loader / Reflective Mapping",
            "score": 45,
            "matches": {
                **match_apis(imports, {"GetProcAddress": 25, "LoadLibraryA": 20, "LoadLibraryW": 20, "LdrLoadDll": 25, "VirtualProtect": 25})
            }
        })

    timestamp_anomalies = file_info.get("timestamp_anomalies", [])
    for anomaly in timestamp_anomalies:
        total_score += anomaly["score"]
        findings.append(anomaly)

    return total_score, findings


def annotate_timestamp_anomalies(file_infos):
    valid_epochs = []
    for info in file_infos:
        timestamps = info.get("timestamps", {})
        ts = timestamps.get("modified_epoch") or timestamps.get("created_epoch")

        if ts is not None:
            valid_epochs.append(ts)

    if not valid_epochs:
        return

    sorted_epochs = sorted(valid_epochs)
    median = sorted_epochs[len(sorted_epochs) // 2]
    now = time.time()

    for info in file_infos:
        timestamps = info.get("timestamps", {})
        if not timestamps:
            continue

        ts = timestamps.get("modified_epoch") or timestamps.get("created_epoch")
        if ts is None:
            continue

        source = "Modified" if timestamps.get("modified_epoch") is not None else "Created"
        display_ts = timestamps.get("modified") or timestamps.get("created")
        anomalies = []

        if ts > now + 86400:
            anomalies.append({
                "type": f"Future {source} Timestamp",
                "score": 30,
                "matches": {
                    "source": source,
                    "timestamp": display_ts
                }
            })

        if ts < datetime.datetime(2000, 1, 1).timestamp():
            anomalies.append({
                "type": f"Historic {source} Timestamp",
                "score": 20,
                "matches": {
                    "source": source,
                    "timestamp": display_ts
                }
            })

        if abs(ts - median) >= 3 * 31536000 and len(valid_epochs) > 1:
            anomalies.append({
                "type": f"{source} Timestamp Outlier",
                "score": 20,
                "matches": {
                    "source": source,
                    "timestamp": display_ts,
                    "median": datetime.datetime.utcfromtimestamp(median).isoformat() + "Z"
                }
            })

        info["timestamp_anomalies"] = anomalies