main.py

import json
import hcl2
import sys
import csv
from datetime import datetime
import platform
import multiprocessing
import os
import subprocess
import re
from typing import Dict, List, Any

def get_task_tree() -> Dict[str, List[Dict[str, str]]]:
    try:
        result = subprocess.run(["task", "--list"], capture_output=True, text=True, check=True)
        lines = result.stdout.splitlines()
        
        tree = {}
        
        for line in lines:
            line = line.strip()
            if not line.startswith("*"):
                continue
                
            match = re.match(r"\*\s+(.*?):\s+(.*)", line)
            
            if match:
                full_name = match.group(1).strip()
                description = match.group(2).strip()
                
                if ":" in full_name:
                    parts = full_name.split(":", 1)
                    namespace = parts[0]
                    task_display_name = parts[1]
                else:
                    namespace = "General"
                    task_display_name = full_name
                
                if namespace not in tree:
                    tree[namespace] = []
                    
                tree[namespace].append({
                    "full_name": full_name,
                    "name": task_display_name,
                    "description": description
                })
        
        return tree
    except Exception as e:
        return {"Error": [{"full_name": "error", "description": str(e)}]}
def get_total_ram_gb():
    try:
        total_bytes = os.sysconf('SC_PAGE_SIZE') * os.sysconf('SC_PHYS_PAGES')
        return round(total_bytes / (1024.**3), 1)
    except ValueError:
        return "N/A"



# Path setup for internal modules
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.append(ROOT_DIR)

from scripts.logger import log, group_start, group_end

def load_json(path: str) -> dict:
    if not os.path.exists(path):
        log.warning(f"File not found: {path}")
        return {}
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception as e:
        log.error(f"Failed to parse JSON at {path}: {str(e)}")
        return {}

def load_hcl(path: str) -> dict:
    if not os.path.exists(path):
        log.warning(f"File not found: {path}")
        return {}
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = hcl2.load(f)
        return {
            key: value.get("default")
            for variable in data.get("variable", [])
            for key, value in variable.items()
        }
    except Exception as e:
        log.error(f"Failed to parse HCL at {path}: {str(e)}")
        return {}

def define_env(env):
    @env.filter
    def from_json(value):
        return json.loads(value)
    
    @env.filter
    def to_datetime(date_str):
        try:
            clean_date = date_str.replace('Z', '+00:00').split('.')[0]
            return datetime.fromisoformat(clean_date)
        except:
            return datetime.now()
    
    group_start("Taskfile Tree Generation")
    task_tree = get_task_tree()
    env.variables.update({
        "task_tree": task_tree})
    group_end()

    group_start("MkDocs Data Injection")
    
    reports_dir = os.path.join(ROOT_DIR, "reports")
    csv_path_throughput = os.path.join(reports_dir, "results.csv")
    csv_path_signatures = os.path.join(reports_dir, "signatures.csv")
    env.variables.update ({"distroless_report_FIPS_validation": load_json(os.path.join(reports_dir, "report_distroless.json"))})
    env.variables.update ({"standard_report_FIPS_validation": load_json(os.path.join(reports_dir, "report_standard.json"))})
    env.variables.update({"kics_report": load_json(os.path.join(reports_dir, "kics-report.json"))})

    throughput_values = []
    signature_values = []
    env.variables.update({"standard_compliance":{"docker_cis": load_json(os.path.join(reports_dir, "standard-docker-cis_full.json")),
                         "k8s_nsa": load_json(os.path.join(reports_dir, "standard-k8s-nsa_full.json")),
                         "k8s_pss_restricted": load_json(os.path.join(reports_dir, "standard-k8s-pss-restricted_full.json")),
                         "security_scan": load_json(os.path.join(reports_dir, "standard-security-full.json"))}})
    env.variables.update({"distroless_compliance":{"docker_cis": load_json(os.path.join(reports_dir, "distroless-docker-cis_full.json")),
                         "k8s_nsa": load_json(os.path.join(reports_dir, "distroless-k8s-nsa_full.json")),
                         "k8s_pss_restricted": load_json(os.path.join(reports_dir, "distroless-k8s-pss-restricted_full.json")),
                         "security_scan": load_json(os.path.join(reports_dir, "distroless-security-full.json"))}})
    env.variables.update({"development_compliance":{"docker_cis": load_json(os.path.join(reports_dir, "development-docker-cis_full.json")),
                         "k8s_nsa": load_json(os.path.join(reports_dir, "development-k8s-nsa_full.json")),
                         "k8s_pss_restricted": load_json(os.path.join(reports_dir, "development-k8s-pss-restricted_full.json")),
                         "security_scan": load_json(os.path.join(reports_dir, "development-security-full.json"))}})
    if os.path.exists(csv_path_throughput):
        try:
            with open(csv_path_throughput, encoding='utf-8') as f:
                throughput_values = list(csv.DictReader(f))
        except Exception as e:
            log.error(f"Failed to read results.csv: {str(e)}")

    if os.path.exists(csv_path_signatures):
        try:
            with open(csv_path_signatures, encoding='utf-8') as f:
                signature_values = list(csv.DictReader(f))
        except Exception as e:
            log.error(f"Failed to read signatures.csv: {str(e)}")
    
    # 2. Configuration & Statistics
    bake_vars = load_hcl(os.path.join(ROOT_DIR, "docker-bake.hcl"))
    version_vars = load_hcl(os.path.join(ROOT_DIR, "versions.hcl"))
    summary = load_json(os.path.join(ROOT_DIR, "reports", "summary.json"))
    raw_benchmarks = load_json(os.path.join(ROOT_DIR, "reports", "benchmark_data.json"))

    # 3. Supply Chain Metadata
    distroless_meta = load_json(os.path.join(reports_dir, "distroless_attestation_details.json"))
    standard_meta = load_json(os.path.join(reports_dir, "standard_attestation_details.json"))
    dev_meta = load_json(os.path.join(reports_dir, "development_attestation_details.json"))

    # 4. Processing Test Stats
    stats = summary.get("summary", {}).get(
        "statistic",
        {"passed": 0, "failed": 0, "broken": 0, "total": 0}
    )

    # 5. Benchmark Fallback (prevents template errors if JSON is missing)
    if not raw_benchmarks or "metrics" not in raw_benchmarks:
        log.warning("Benchmark JSON missing. Injecting empty fallback structure.")
        benchmarks = {
            "metadata": {"fips": {"openssl_version": "N/A"}},
            "metrics": {
                "AES-256-GCM": {"fips": [0]*6, "ubuntu": [0]*6, "debian": [0]*6, "alpine": [0]*6},
                "sha256": {"fips": [0]*6, "ubuntu": [0]*6, "debian": [0]*6, "alpine": [0]*6},
                "sha512": {"fips": [0]*6, "ubuntu": [0]*6, "debian": [0]*6, "alpine": [0]*6},
                "sha3-256": {"fips": [0]*6, "ubuntu": [0]*6, "debian": [0]*6, "alpine": [0]*6}
            }
        }
    else:
        benchmarks = raw_benchmarks

    owner = bake_vars.get("OWNER", "taha2samy")
    repo = bake_vars.get("REPO_NAME", "wolfi-openssl-fips")

    # 6. Global Variable Injection
    env.variables.update({
        "project_name": os.getenv("PROJECT_NAME", "Wolfi OpenSSL FIPS"),
        "owner": owner,
        "repo_name": repo,
        "registry": bake_vars.get("REGISTRY", "ghcr.io"),
        "core_version": bake_vars.get("CORE_VERSION", "3.5.5"),
        "fips_version": bake_vars.get("FIPS_VERSION", "3.1.2"),
        "base_image": version_vars.get("BASE_IMAGE", "N/A"),
        "static_image": version_vars.get("STATIC_IMAGE", "N/A"),
        "generation_date": os.getenv("GENERATION_DATE", datetime.utcnow().strftime("%Y-%m-%d")),
        "test_stats": stats,
        "test_badge_color": "red" if (stats["failed"] + stats["broken"]) > 0 else "brightgreen",
        "benchmark_data": benchmarks,
        "bench_results_raw": json.dumps(throughput_values),
        "bench_signatures_raw": json.dumps(signature_values)
    })

    # 7. Dependency List
    env.variables["packages"] = [
        {"name": key.replace("_VER", "").replace("_", " ").title(), "version": value}
        for key, value in version_vars.items()
        if key not in {"BASE_IMAGE", "STATIC_IMAGE"}
    ]

    # 8. Provenance & SBOM URLs
    env.variables.update({
        "distroless_digest": distroless_meta.get("provenance", {}).get("digest", "N/A"),
        "distroless_url": distroless_meta.get("provenance", {}).get("url", "#"),
        "distroless_sbom_url": distroless_meta.get("sbom", {}).get("url", "#"),
        "standard_digest": standard_meta.get("provenance", {}).get("digest", "N/A"),
        "standard_url": standard_meta.get("provenance", {}).get("url", "#"),
        "standard_sbom_url": standard_meta.get("sbom", {}).get("url", "#"),
        "dev_digest": dev_meta.get("provenance", {}).get("digest", "N/A"),
        "dev_url": dev_meta.get("provenance", {}).get("url", "#"),
        "dev_sbom_url": dev_meta.get("sbom", {}).get("url", "#"),
    })
    env.variables["hardware_context"] = {
        "system": platform.system(),                # e.g., Linux
        "release": platform.release(),              # Kernel version
        "architecture": platform.machine(),         # e.g., x86_64
        "cpu_cores": multiprocessing.cpu_count(),   # Number of logical cores
        "ram_gb": get_total_ram_gb(),               # Total RAM in GB
        "runner": os.getenv("CI", "Local Machine")  # Checks if running in CI
    }
    env.variables["github"] = {
        "repository": os.getenv("GITHUB_REPOSITORY", f"{owner}/{repo}"),
        "commit_sha": os.getenv("GITHUB_SHA", "0000000"),
        "run_id": os.getenv("GITHUB_RUN_ID", "#"),
    }

    @env.macro
    def render_test_status(status: str) -> str:
        icons = {"passed": "✅", "failed": "❌", "broken": "💥"}
        return icons.get(status, "❓")

    log.info("Documentation context successfully injected.")
    group_end()