Prowler Compliance

#!/usr/bin/env python3
"""
Cloud-auditor pass template for any Prowler compliance framework.

Encode explicit REPLACE decisions per (requirement_id, provider) pair below.
Each decision FULLY overwrites the legacy Checks list for that requirement.

Workflow:
    1. Run build_inventory.py first to cache per-provider check metadata.
    2. Run dump_section.py to see current mappings for the catalog you're auditing.
    3. Fill in DECISIONS below with explicit check lists.
    4. Run this script — it pre-validates every check id against the inventory
       and aborts with stderr listing typos before writing.

Decision rules (apply as a hostile cloud auditor):
    - The Prowler check's title/risk MUST literally describe what the AR text says.
      "Related" is not enough.
    - If no check actually addresses the requirement, leave `[]` (= MANUAL).
      HONEST MANUAL is worth more than padded coverage.
    - Missing provider key = leave the legacy mapping untouched.
    - Empty list `[]` = explicitly MANUAL (overwrites legacy).

Usage:
    # 1. Copy this file to /tmp/audit_<framework>.py and fill in DECISIONS
    # 2. Edit FRAMEWORK_KEY below to match your framework file naming
    # 3. Run:
    python /tmp/audit_<framework>.py
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

# ---------------------------------------------------------------------------
# Configure for your framework
# ---------------------------------------------------------------------------

# Framework file basename inside prowler/compliance/{provider}/.
# If your framework is called "cis_5.0_aws.json", FRAMEWORK_KEY is "cis_5.0".
# If the file is "ccc_aws.json", FRAMEWORK_KEY is "ccc".
FRAMEWORK_KEY = "ccc"

# Which providers to apply decisions to.
PROVIDERS = ["aws", "azure", "gcp"]

PROWLER_DIR = Path("prowler/compliance")
CHECK_INV = {prov: Path(f"/tmp/checks_{prov}.json") for prov in PROVIDERS}


# ---------------------------------------------------------------------------
# DECISIONS — encode one entry per requirement you want to audit
# ---------------------------------------------------------------------------

# DECISIONS[requirement_id][provider] = list[str] of check ids
# See SKILL.md → "Audit Reference Table: Requirement Text → Prowler Checks"
# for a comprehensive mapping cheat sheet built from a 172-AR CCC audit.

DECISIONS: dict[str, dict[str, list[str]]] = {}

# ---- Example entries (delete and replace with your own) ----

# Example 1: TLS in transit enforced (non-SSH traffic)
# DECISIONS["CCC.Core.CN01.AR01"] = {
#     "aws": [
#         "cloudfront_distributions_https_enabled",
#         "cloudfront_distributions_origin_traffic_encrypted",
#         "s3_bucket_secure_transport_policy",
#         "elbv2_ssl_listeners",
#         "rds_instance_transport_encrypted",
#         "kafka_cluster_in_transit_encryption_enabled",
#         "redshift_cluster_in_transit_encryption_enabled",
#         "opensearch_service_domains_https_communications_enforced",
#     ],
#     "azure": [
#         "storage_secure_transfer_required_is_enabled",
#         "app_minimum_tls_version_12",
#         "postgresql_flexible_server_enforce_ssl_enabled",
#         "sqlserver_recommended_minimal_tls_version",
#     ],
#     "gcp": [
#         "cloudsql_instance_ssl_connections",
#     ],
# }

# Example 2: MANUAL — no Prowler check exists
# DECISIONS["CCC.Core.CN01.AR07"] = {
#     "aws": [],    # no IANA port/protocol check exists in Prowler
#     "azure": [],
#     "gcp": [],
# }

# Example 3: Reuse a decision for multiple sibling ARs
# DECISIONS["CCC.ObjStor.CN05.AR02"] = DECISIONS["CCC.ObjStor.CN05.AR01"]


# ---------------------------------------------------------------------------
# Driver — do not edit below
# ---------------------------------------------------------------------------

def load_inventory(provider: str) -> dict:
    path = CHECK_INV[provider]
    if not path.exists():
        raise SystemExit(
            f"Check inventory missing: {path}\n"
            f"Run: python skills/prowler-compliance/assets/build_inventory.py {provider}"
        )
    with open(path) as f:
        return json.load(f)


def resolve_json_path(provider: str) -> Path:
    """Resolve the JSON file path for a given provider.

    Handles both shapes: {FRAMEWORK_KEY}_{provider}.json (ccc_aws.json) and
    cases where FRAMEWORK_KEY already contains the provider suffix.
    """
    candidates = [
        PROWLER_DIR / provider / f"{FRAMEWORK_KEY}_{provider}.json",
        PROWLER_DIR / provider / f"{FRAMEWORK_KEY}.json",
    ]
    for c in candidates:
        if c.exists():
            return c
    raise SystemExit(
        f"Could not find framework JSON for provider={provider} "
        f"with FRAMEWORK_KEY={FRAMEWORK_KEY}. Tried: {candidates}"
    )


def plan_for_provider(
    provider: str,
) -> tuple[Path, dict, tuple[int, int, int], list[tuple[str, str]]]:
    """Build the updated JSON for one provider without writing it.

    Returns (path, mutated_data, (touched, added, removed), unknowns).
    Writing is deferred to a second pass so that a typo in any provider
    aborts the whole run before any file on disk changes.
    """
    path = resolve_json_path(provider)
    with open(path) as f:
        data = json.load(f)
    inv = load_inventory(provider)

    touched = 0
    add_count = 0
    rm_count = 0
    unknown: list[tuple[str, str]] = []

    for req in data["Requirements"]:
        rid = req["Id"]
        if rid not in DECISIONS or provider not in DECISIONS[rid]:
            continue
        new_checks = list(dict.fromkeys(DECISIONS[rid][provider]))
        for c in new_checks:
            if c not in inv:
                unknown.append((rid, c))
        before = set(req.get("Checks") or [])
        after = set(new_checks)
        rm_count += len(before - after)
        add_count += len(after - before)
        req["Checks"] = new_checks
        touched += 1

    return path, data, (touched, add_count, rm_count), unknown


def main() -> int:
    if not DECISIONS:
        print("No DECISIONS encoded. Fill in the DECISIONS dict and re-run.")
        return 1
    print(f"Applying {len(DECISIONS)} decisions to framework '{FRAMEWORK_KEY}'...")

    # Pass 1: validate every provider before touching disk. A typo in any
    # provider must abort the run before ANY file has been rewritten.
    plans: list[tuple[str, Path, dict, tuple[int, int, int]]] = []
    all_unknown: list[tuple[str, str, str]] = []
    for provider in PROVIDERS:
        path, data, counts, unknown = plan_for_provider(provider)
        for rid, c in unknown:
            all_unknown.append((provider, rid, c))
        plans.append((provider, path, data, counts))

    if all_unknown:
        print("\n!! UNKNOWN CHECK IDS (typos?):", file=sys.stderr)
        for provider, rid, c in all_unknown:
            print(f"   {provider} {rid} -> {c}", file=sys.stderr)
        print(
            "\nAborting: fix the check ids above and re-run. "
            "No files were modified.",
            file=sys.stderr,
        )
        return 2

    # Pass 2: all providers validated cleanly — write.
    for provider, path, data, (touched, added, removed) in plans:
        with open(path, "w") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
            f.write("\n")
        print(
            f"  {provider}: touched={touched} added={added} removed={removed}"
        )
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
Build a per-provider check inventory by scanning Prowler's check metadata files.

Outputs one JSON per provider at /tmp/checks_{provider}.json with the shape:
    {
        "check_id": {
            "service": "...",
            "subservice": "...",
            "resource": "...",
            "severity": "...",
            "title": "...",
            "description": "...",
            "risk": "..."
        },
        ...
    }

This is the reference used by audit_framework_template.py for pre-validation
(every check id in the audit ledger must exist in the inventory) and by
query_checks.py for keyword/service lookup.

Usage:
    python skills/prowler-compliance/assets/build_inventory.py
    # Or for a specific provider:
    python skills/prowler-compliance/assets/build_inventory.py aws

Output:
    /tmp/checks_{provider}.json for every provider discovered under
    prowler/providers/ with a services/ directory.
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

PROVIDERS_ROOT = Path("prowler/providers")


def discover_providers() -> list[str]:
    """Return every provider that currently has a services/ directory.

    Derived from the filesystem so new providers are picked up automatically
    and stale hard-coded lists cannot drift from the repo.
    """
    if not PROVIDERS_ROOT.exists():
        return []
    return sorted(
        p.name
        for p in PROVIDERS_ROOT.iterdir()
        if p.is_dir() and (p / "services").is_dir()
    )


def build_for_provider(provider: str) -> dict:
    inventory: dict[str, dict] = {}
    base = Path(f"prowler/providers/{provider}/services")
    if not base.exists():
        print(f"  skip {provider}: no services directory", file=sys.stderr)
        return inventory
    for meta_path in base.rglob("*.metadata.json"):
        try:
            with open(meta_path) as f:
                data = json.load(f)
        except Exception as exc:
            print(f"  warn: cannot parse {meta_path}: {exc}", file=sys.stderr)
            continue
        cid = data.get("CheckID") or meta_path.stem.replace(".metadata", "")
        inventory[cid] = {
            "service": data.get("ServiceName", ""),
            "subservice": data.get("SubServiceName", ""),
            "resource": data.get("ResourceType", ""),
            "severity": data.get("Severity", ""),
            "title": data.get("CheckTitle", ""),
            "description": data.get("Description", ""),
            "risk": data.get("Risk", ""),
        }
    return inventory


def main() -> int:
    providers = sys.argv[1:] or discover_providers()
    if not providers:
        print(
            f"error: no providers found under {PROVIDERS_ROOT}/",
            file=sys.stderr,
        )
        return 1
    for provider in providers:
        inv = build_for_provider(provider)
        out_path = Path(f"/tmp/checks_{provider}.json")
        with open(out_path, "w") as f:
            json.dump(inv, f, indent=2)
        print(f"  {provider}: {len(inv)} checks → {out_path}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

{
  "Framework": "CIS",
  "Name": "CIS Amazon Web Services Foundations Benchmark v5.0.0",
  "Version": "5.0",
  "Provider": "AWS",
  "Description": "The CIS Amazon Web Services Foundations Benchmark provides prescriptive guidance for configuring security options for a subset of Amazon Web Services with an emphasis on foundational, testable, and architecture agnostic settings.",
  "Requirements": [
    {
      "Id": "1.1",
      "Description": "Maintain current contact details",
      "Checks": [
        "account_maintain_current_contact_details"
      ],
      "Attributes": [
        {
          "Section": "1 Identity and Access Management",
          "Profile": "Level 1",
          "AssessmentStatus": "Manual",
          "Description": "Ensure contact email and telephone details for AWS accounts are current and map to more than one individual in your organization.",
          "RationaleStatement": "If an AWS account is observed to be behaving in a prohibited or suspicious manner, AWS will attempt to contact the account owner by email and phone using the contact details listed. If this is unsuccessful and the account behavior is not corrected then AWS may suspend the account.",
          "ImpactStatement": "",
          "RemediationProcedure": "This activity can only be performed via the AWS Console. Navigate to Account Settings and update contact information.",
          "AuditProcedure": "This activity can only be performed via the AWS Console. Navigate to Account Settings and verify contact information is current.",
          "AdditionalInformation": "",
          "DefaultValue": "",
          "References": "https://docs.aws.amazon.com/accounts/latest/reference/manage-acct-update-contact.html"
        }
      ]
    },
    {
      "Id": "1.2",
      "Description": "Ensure security contact information is registered",
      "Checks": [
        "account_security_contact_information_is_registered"
      ],
      "Attributes": [
        {
          "Section": "1 Identity and Access Management",
          "Profile": "Level 1",
          "AssessmentStatus": "Automated",
          "Description": "AWS provides customers with the option to specify the contact information for the account's security team. It is recommended that this information be provided.",
          "RationaleStatement": "Specifying security-specific contact information will help ensure that security advisories sent by AWS reach the team in your organization that is best equipped to respond to them.",
          "ImpactStatement": "",
          "RemediationProcedure": "Navigate to AWS Console > Account > Alternate Contacts and add security contact information.",
          "AuditProcedure": "Run: aws account get-alternate-contact --alternate-contact-type SECURITY",
          "AdditionalInformation": "",
          "DefaultValue": "By default, no security contact is registered.",
          "References": "https://docs.aws.amazon.com/accounts/latest/reference/manage-acct-update-contact-alternate.html"
        }
      ]
    },
    {
      "Id": "1.3",
      "Description": "Ensure no 'root' user account access key exists",
      "Checks": [
        "iam_no_root_access_key"
      ],
      "Attributes": [
        {
          "Section": "1 Identity and Access Management",
          "Profile": "Level 1",
          "AssessmentStatus": "Automated",
          "Description": "The 'root' user account is the most privileged user in an AWS account. AWS Access Keys provide programmatic access to a given AWS account. It is recommended that all access keys associated with the 'root' user account be deleted.",
          "RationaleStatement": "Deleting access keys associated with the 'root' user account limits vectors by which the account can be compromised. Additionally, deleting the root access keys encourages the creation and use of role based accounts that are least privileged.",
          "ImpactStatement": "",
          "RemediationProcedure": "Navigate to IAM console, select root user, Security credentials tab, and delete any access keys.",
          "AuditProcedure": "Run: aws iam get-account-summary | grep 'AccountAccessKeysPresent'",
          "AdditionalInformation": "IAM User account root for us-gov cloud regions is not enabled by default.",
          "DefaultValue": "By default, no root access keys exist.",
          "References": "https://docs.aws.amazon.com/general/latest/gr/aws-access-keys-best-practices.html"
        }
      ]
    },
    {
      "Id": "1.4",
      "Description": "Ensure MFA is enabled for the 'root' user account",
      "Checks": [
        "iam_root_mfa_enabled"
      ],
      "Attributes": [
        {
          "Section": "1 Identity and Access Management",
          "Profile": "Level 1",
          "AssessmentStatus": "Automated",
          "Description": "The 'root' user account is the most privileged user in an AWS account. Multi-factor Authentication (MFA) adds an extra layer of protection on top of a username and password. With MFA enabled, when a user signs in to an AWS website, they will be prompted for their username and password as well as for an authentication code from their AWS MFA device.",
          "RationaleStatement": "Enabling MFA provides increased security for console access as it requires the authenticating principal to possess a device that emits a time-sensitive key and have knowledge of a credential.",
          "ImpactStatement": "",
          "RemediationProcedure": "Using IAM console, navigate to Dashboard and choose Activate MFA on your root account.",
          "AuditProcedure": "Run: aws iam get-account-summary | grep 'AccountMFAEnabled'. Ensure the value is 1.",
          "AdditionalInformation": "",
          "DefaultValue": "MFA is not enabled by default.",
          "References": "https://docs.aws.amazon.com/IAM/latest/UserGuide/id_root-user.html#id_root-user_manage_mfa"
        }
      ]
    },
    {
      "Id": "1.5",
      "Description": "Ensure hardware MFA is enabled for the 'root' user account",
      "Checks": [
        "iam_root_hardware_mfa_enabled"
      ],
      "Attributes": [
        {
          "Section": "1 Identity and Access Management",
          "Profile": "Level 2",
          "AssessmentStatus": "Automated",
          "Description": "The 'root' user account is the most privileged user in an AWS account. MFA adds an extra layer of protection on top of a user name and password. With MFA enabled, when a user signs in to an AWS website, they will be prompted for their user name and password as well as for an authentication code from their AWS MFA device. For Level 2, it is recommended that the root user account be protected with a hardware MFA.",
          "RationaleStatement": "A hardware MFA has a smaller attack surface than a virtual MFA. For example, a hardware MFA does not suffer from the attack surface introduced by the mobile smartphone on which a virtual MFA resides.",
          "ImpactStatement": "Using a hardware MFA device instead of a virtual MFA may result in additional hardware costs.",
          "RemediationProcedure": "Using IAM console, navigate to Dashboard, select root user, and configure hardware MFA device.",
          "AuditProcedure": "Run: aws iam list-virtual-mfa-devices and verify the root account is not using a virtual MFA.",
          "AdditionalInformation": "For recommendations on protecting hardware MFA devices, refer to https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_mfa_lost-or-broken.html",
          "DefaultValue": "MFA is not enabled by default.",
          "References": "https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_mfa_enable_physical.html"
        }
      ]
    },
    {
      "Id": "2.1.1",
      "Description": "Ensure S3 Bucket Policy is set to deny HTTP requests",
      "Checks": [
        "s3_bucket_secure_transport_policy"
      ],
      "Attributes": [
        {
          "Section": "2 Storage",
          "SubSection": "2.1 Simple Storage Service (S3)",
          "Profile": "Level 2",
          "AssessmentStatus": "Automated",
          "Description": "At the Amazon S3 bucket level, you can configure permissions through a bucket policy making the objects accessible only through HTTPS.",
          "RationaleStatement": "By default, Amazon S3 allows both HTTP and HTTPS requests. To achieve only allowing access to Amazon S3 objects through HTTPS you also have to explicitly deny access to HTTP requests. Bucket policies that allow HTTPS requests without explicitly denying HTTP requests will not comply with this recommendation.",
          "ImpactStatement": "Enabling this setting will result in rejection of requests that do not use HTTPS for S3 bucket operations.",
          "RemediationProcedure": "Add a bucket policy with condition aws:SecureTransport: false that denies all s3 actions.",
          "AuditProcedure": "Review bucket policies for Deny statements with aws:SecureTransport: false condition.",
          "AdditionalInformation": "",
          "DefaultValue": "By default, S3 buckets allow both HTTP and HTTPS requests.",
          "References": "https://aws.amazon.com/blogs/security/how-to-use-bucket-policies-and-apply-defense-in-depth-to-help-secure-your-amazon-s3-data/"
        }
      ]
    }
  ]
}

# FINOS Common Cloud Controls (CCC) sync config for sync_framework.py.
#
# Usage:
#   python skills/prowler-compliance/assets/sync_framework.py \
#          skills/prowler-compliance/assets/configs/ccc.yaml
#
# Prerequisite: run the upstream fetch step from SKILL.md Workflow A Step 1 to
# populate upstream.dir with the raw FINOS catalog YAML files.

framework:
  name: CCC
  display_name: Common Cloud Controls Catalog (CCC)
  version: v2025.10
  # The {provider_display} placeholder is replaced at output time with the
  # per-provider display string from the providers list below.
  description_template: "Common Cloud Controls Catalog (CCC) for {provider_display}"

providers:
  - key: aws
    display: AWS
  - key: azure
    display: Azure
  - key: gcp
    display: GCP

output:
  # Supported placeholders: {provider}, {framework}, {version}.
  # For versioned frameworks like CIS the template would be
  # "prowler/compliance/{provider}/cis_{version}_{provider}.json".
  path_template: "prowler/compliance/{provider}/ccc_{provider}.json"

upstream:
  # Directory containing the cached FINOS catalog YAMLs. Populate via
  # SKILL.md Workflow A Step 1 (gh api raw download commands).
  dir: /tmp/ccc_upstream
  fetch_docs: "See SKILL.md Workflow A Step 1 for gh api fetch commands"

parser:
  # Name of the parser module under parsers/ (loaded dynamically by the
  # runner). For FINOS CCC YAML this is always finos_ccc.
  module: finos_ccc

  # FINOS CCC catalog files in load order. Core first so its ARs render
  # first in the output JSON.
  catalog_files:
    - core_ccc.yaml
    - management_auditlog.yaml
    - management_logging.yaml
    - management_monitoring.yaml
    - storage_object.yaml
    - networking_loadbalancer.yaml
    - networking_vpc.yaml
    - crypto_key.yaml
    - crypto_secrets.yaml
    - database_warehouse.yaml
    - database_vector.yaml
    - database_relational.yaml
    - devtools_build.yaml
    - devtools_container-registry.yaml
    - identity_iam.yaml
    - ai-ml_gen-ai.yaml
    - ai-ml_mlde.yaml
    - app-integration_message.yaml
    - compute_serverless-computing.yaml

  # Shape-2 catalogs (storage/object) reference the family via id only
  # (e.g. "CCC.ObjStor.Data") with no human-readable title or description
  # in the YAML. Map the suffix (after the last dot) to a canonical title
  # and description so the generated JSON has consistent FamilyName fields
  # regardless of upstream shape.
  family_id_title:
    Data: Data
    IAM: Identity and Access Management
    Identity: Identity and Access Management
    Encryption: Encryption
    Logging: Logging and Monitoring
    Network: Network Security
    Availability: Availability
    Integrity: Integrity
    Confidentiality: Confidentiality
  family_id_description:
    Data: "The Data control family ensures the confidentiality, integrity, availability, and sovereignty of data across its lifecycle."
    IAM: "The Identity and Access Management control family ensures that only trusted and authenticated entities can access resources."

post_processing:
  # Collapse FamilyName variants that appear inconsistently across upstream
  # catalogs. The Prowler UI groups by Attributes[0].FamilyName exactly,
  # so each variant would otherwise become a separate tree branch.
  family_name_normalization:
    "Logging & Monitoring": "Logging and Monitoring"
    "Logging and Metrics Publication": "Logging and Monitoring"

  # Preserve existing Checks lists from the legacy Prowler JSON when
  # regenerating. The runner builds two lookup tables from the legacy
  # output: a primary index by Id, and fallback indexes composed of
  # attribute field names.
  #
  # primary_key: the top-level requirement field to use as the primary
  #   lookup key (almost always "Id")
  # fallback_keys: a list of composite keys. Each composite key is a list
  #   of Attributes[0] field names to join into a tuple. List-valued fields
  #   (like Applicability) are frozen to frozenset so the tuple is hashable.
  #
  # CCC uses (Section, Applicability) because Applicability is a CCC-only
  # top-level attribute field. CIS would use (Section, Profile). NIST would
  # use (ItemId,). The fallback is how renumbered or rewritten ids still
  # recover their check mappings.
  #
  # legacy_path_template (optional): path to read legacy Checks FROM.
  # Defaults to output.path_template, which is correct for unversioned
  # frameworks (like CCC) where regeneration overwrites the same file.
  # For versioned frameworks that write to a new file on each version
  # bump (e.g. cis_5.1_aws.json while the legacy mappings live in
  # cis_5.0_aws.json), set this to the previous-version path so Checks
  # are preserved instead of lost:
  #   legacy_path_template: "prowler/compliance/{provider}/cis_5.0_{provider}.json"
  check_preservation:
    primary_key: Id
    fallback_keys:
      - [Section, Applicability]

#!/usr/bin/env python3
"""
Dump every requirement of a compliance framework for a given id prefix across
providers, with their current Check mappings.

Useful for reviewing a whole control family in one pass before encoding audit
decisions in audit_framework_template.py.

Usage:
    # Dump all CCC.Core requirements across aws/azure/gcp
    python skills/prowler-compliance/assets/dump_section.py ccc "CCC.Core."

    # Dump all CIS 5.0 section 1 requirements for AWS only
    python skills/prowler-compliance/assets/dump_section.py cis_5.0_aws "1."

Arguments:
    framework_key: file prefix inside prowler/compliance/{provider}/ without
                   the provider suffix. Examples:
                     - "ccc"        → loads ccc_aws.json / ccc_azure.json / ccc_gcp.json
                     - "cis_5.0_aws" → loads only that one file
                     - "iso27001_2022" → loads all providers
    id_prefix:     Requirement id prefix to filter by (e.g. "CCC.Core.",
                   "1.1.", "A.5.").
"""
from __future__ import annotations

import json
import sys
from collections import defaultdict
from pathlib import Path

PROWLER_COMPLIANCE_DIR = Path("prowler/compliance")


def main() -> int:
    if len(sys.argv) < 3:
        print(__doc__)
        return 1

    framework_key = sys.argv[1]
    id_prefix = sys.argv[2]

    # Find matching JSON files across all providers
    candidates: list[tuple[str, Path]] = []
    for prov_dir in sorted(PROWLER_COMPLIANCE_DIR.iterdir()):
        if not prov_dir.is_dir():
            continue
        for json_path in prov_dir.glob("*.json"):
            stem = json_path.stem
            if stem == framework_key or stem.startswith(f"{framework_key}_") \
                    or stem == f"{framework_key}_{prov_dir.name}":
                candidates.append((prov_dir.name, json_path))

    if not candidates:
        print(f"No files matching '{framework_key}'", file=sys.stderr)
        return 2

    discovered_providers = sorted({prov for prov, _ in candidates})

    by_id: dict[str, dict] = defaultdict(dict)
    for prov, path in candidates:
        with open(path) as f:
            data = json.load(f)
        for req in data["Requirements"]:
            if req["Id"].startswith(id_prefix):
                by_id[req["Id"]][prov] = {
                    "desc": req.get("Description", ""),
                    "sec": (req.get("Attributes") or [{}])[0].get("Section", ""),
                    "obj": (req.get("Attributes") or [{}])[0].get(
                        "SubSectionObjective", ""
                    ),
                    "checks": req.get("Checks") or [],
                }

    for ar_id in sorted(by_id):
        rows = by_id[ar_id]
        sample = next(iter(rows.values()))
        print(f"\n### {ar_id}")
        print(f"  desc: {sample['desc']}")
        if sample["sec"]:
            print(f"  sec : {sample['sec']}")
        if sample["obj"]:
            print(f"  obj : {sample['obj']}")
        for prov in discovered_providers:
            if prov in rows:
                checks = rows[prov]["checks"]
                print(f"  {prov}: ({len(checks)}) {checks}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

{
  "Framework": "ENS",
  "Name": "ENS RD 311/2022 - Categoria Alta",
  "Version": "RD2022",
  "Provider": "AWS",
  "Description": "The accreditation scheme of the ENS (Esquema Nacional de Seguridad - National Security Scheme of Spain) has been developed by the Ministry of Finance and Public Administrations and the CCN (National Cryptological Center). This includes the basic principles and minimum requirements necessary for the adequate protection of information.",
  "Requirements": [
    {
      "Id": "op.acc.1.aws.iam.2",
      "Description": "Proveedor de identidad centralizado",
      "Attributes": [
        {
          "IdGrupoControl": "op.acc.1",
          "Marco": "operacional",
          "Categoria": "control de acceso",
          "DescripcionControl": "Es muy recomendable la utilizacion de un proveedor de identidades que permita administrar las identidades en un lugar centralizado, en vez de utilizar IAM para ello.",
          "Nivel": "alto",
          "Tipo": "requisito",
          "Dimensiones": [
            "trazabilidad",
            "autenticidad"
          ],
          "ModoEjecucion": "automatico",
          "Dependencias": []
        }
      ],
      "Checks": [
        "iam_check_saml_providers_sts"
      ]
    },
    {
      "Id": "op.acc.2.aws.iam.4",
      "Description": "Requisitos de acceso",
      "Attributes": [
        {
          "IdGrupoControl": "op.acc.2",
          "Marco": "operacional",
          "Categoria": "control de acceso",
          "DescripcionControl": "Se debera delegar en cuentas administradoras la administracion de la organizacion, dejando la cuenta maestra sin uso y con las medidas de seguridad pertinentes.",
          "Nivel": "alto",
          "Tipo": "requisito",
          "Dimensiones": [
            "confidencialidad",
            "integridad",
            "trazabilidad",
            "autenticidad"
          ],
          "ModoEjecucion": "automatico",
          "Dependencias": []
        }
      ],
      "Checks": [
        "iam_avoid_root_usage"
      ]
    },
    {
      "Id": "op.acc.3.r1.aws.iam.1",
      "Description": "Segregacion rigurosa",
      "Attributes": [
        {
          "IdGrupoControl": "op.acc.3.r1",
          "Marco": "operacional",
          "Categoria": "control de acceso",
          "DescripcionControl": "En caso de ser de aplicacion, la segregacion debera tener en cuenta la separacion de las funciones de configuracion y mantenimiento y de auditoria de cualquier otra.",
          "Nivel": "alto",
          "Tipo": "refuerzo",
          "Dimensiones": [
            "confidencialidad",
            "integridad",
            "trazabilidad",
            "autenticidad"
          ],
          "ModoEjecucion": "automatico",
          "Dependencias": []
        }
      ],
      "Checks": [
        "iam_support_role_created"
      ]
    },
    {
      "Id": "op.exp.8.aws.cloudwatch.1",
      "Description": "Registro de la actividad",
      "Attributes": [
        {
          "IdGrupoControl": "op.exp.8",
          "Marco": "operacional",
          "Categoria": "explotacion",
          "DescripcionControl": "Se registraran las actividades de los usuarios en el sistema, de forma que se pueda identificar que acciones ha realizado cada usuario.",
          "Nivel": "medio",
          "Tipo": "requisito",
          "Dimensiones": [
            "trazabilidad"
          ],
          "ModoEjecucion": "automatico",
          "Dependencias": []
        }
      ],
      "Checks": [
        "cloudtrail_multi_region_enabled",
        "cloudwatch_log_group_retention_policy_specific_days_enabled"
      ]
    },
    {
      "Id": "mp.info.3.aws.s3.1",
      "Description": "Cifrado de la informacion",
      "Attributes": [
        {
          "IdGrupoControl": "mp.info.3",
          "Marco": "medidas de proteccion",
          "Categoria": "proteccion de la informacion",
          "DescripcionControl": "La informacion con un nivel de clasificacion CONFIDENCIAL o superior debera ser cifrada.",
          "Nivel": "bajo",
          "Tipo": "medida",
          "Dimensiones": [
            "confidencialidad"
          ],
          "ModoEjecucion": "automatico",
          "Dependencias": []
        }
      ],
      "Checks": [
        "s3_bucket_default_encryption",
        "s3_bucket_kms_encryption"
      ]
    }
  ]
}

{
  "Framework": "CUSTOM-FRAMEWORK",
  "Name": "Custom Security Framework Example v1.0",
  "Version": "1.0",
  "Provider": "AWS",
  "Description": "This is a template for creating custom compliance frameworks using the generic attribute model. Use this when creating frameworks that don't match existing attribute types (CIS, ISO, ENS, MITRE, etc.).",
  "Requirements": [
    {
      "Id": "SEC-001",
      "Description": "Ensure all storage resources are encrypted at rest",
      "Name": "Storage Encryption",
      "Attributes": [
        {
          "ItemId": "SEC-001",
          "Section": "Data Protection",
          "SubSection": "Encryption",
          "SubGroup": "Storage",
          "Service": "s3",
          "Type": "Automated"
        }
      ],
      "Checks": [
        "s3_bucket_default_encryption",
        "rds_instance_storage_encrypted",
        "ec2_ebs_volume_encryption"
      ]
    },
    {
      "Id": "SEC-002",
      "Description": "Ensure all network traffic is encrypted in transit",
      "Name": "Network Encryption",
      "Attributes": [
        {
          "ItemId": "SEC-002",
          "Section": "Data Protection",
          "SubSection": "Encryption",
          "SubGroup": "Network",
          "Service": "multiple",
          "Type": "Automated"
        }
      ],
      "Checks": [
        "s3_bucket_secure_transport_policy",
        "elb_ssl_listeners",
        "cloudfront_distributions_https_enabled"
      ]
    },
    {
      "Id": "IAM-001",
      "Description": "Ensure MFA is enabled for all privileged accounts",
      "Name": "Multi-Factor Authentication",
      "Attributes": [
        {
          "ItemId": "IAM-001",
          "Section": "Identity and Access Management",
          "SubSection": "Authentication",
          "SubGroup": "MFA",
          "Service": "iam",
          "Type": "Automated"
        }
      ],
      "Checks": [
        "iam_root_mfa_enabled",
        "iam_user_mfa_enabled_console_access"
      ]
    },
    {
      "Id": "LOG-001",
      "Description": "Ensure logging is enabled for all critical services",
      "Name": "Centralized Logging",
      "Attributes": [
        {
          "ItemId": "LOG-001",
          "Section": "Logging and Monitoring",
          "SubSection": "Audit Logs",
          "SubGroup": "CloudTrail",
          "Service": "cloudtrail",
          "Type": "Automated"
        }
      ],
      "Checks": [
        "cloudtrail_multi_region_enabled",
        "cloudtrail_s3_dataevents_read_enabled",
        "cloudtrail_s3_dataevents_write_enabled"
      ]
    },
    {
      "Id": "MANUAL-001",
      "Description": "Ensure security policies are reviewed annually",
      "Name": "Policy Review",
      "Attributes": [
        {
          "ItemId": "MANUAL-001",
          "Section": "Governance",
          "SubSection": "Policy Management",
          "Service": "manual",
          "Type": "Manual"
        }
      ],
      "Checks": []
    }
  ]
}

{
  "Framework": "ISO27001",
  "Name": "ISO/IEC 27001 Information Security Management Standard 2022",
  "Version": "2022",
  "Provider": "AWS",
  "Description": "ISO (the International Organization for Standardization) and IEC (the International Electrotechnical Commission) form the specialized system for worldwide standardization. This framework maps AWS security controls to ISO 27001:2022 requirements.",
  "Requirements": [
    {
      "Id": "A.5.1",
      "Description": "Information security policy and topic-specific policies should be defined, approved by management, published, communicated to and acknowledged by relevant personnel and relevant interested parties, and reviewed at planned intervals and if significant changes occur.",
      "Name": "Policies for information security",
      "Attributes": [
        {
          "Category": "A.5 Organizational controls",
          "Objetive_ID": "A.5.1",
          "Objetive_Name": "Policies for information security",
          "Check_Summary": "Verify that information security policies are defined and implemented through security monitoring services."
        }
      ],
      "Checks": [
        "securityhub_enabled",
        "wellarchitected_workload_no_high_or_medium_risks"
      ]
    },
    {
      "Id": "A.5.2",
      "Description": "Information security roles and responsibilities should be defined and allocated according to the organisation needs.",
      "Name": "Roles and Responsibilities",
      "Attributes": [
        {
          "Category": "A.5 Organizational controls",
          "Objetive_ID": "A.5.2",
          "Objetive_Name": "Roles and Responsibilities",
          "Check_Summary": "Verify that IAM roles and responsibilities are properly defined."
        }
      ],
      "Checks": []
    },
    {
      "Id": "A.5.3",
      "Description": "Conflicting duties and conflicting areas of responsibility should be segregated.",
      "Name": "Segregation of Duties",
      "Attributes": [
        {
          "Category": "A.5 Organizational controls",
          "Objetive_ID": "A.5.3",
          "Objetive_Name": "Segregation of Duties",
          "Check_Summary": "Verify that duties are segregated through separate IAM roles."
        }
      ],
      "Checks": [
        "iam_securityaudit_role_created"
      ]
    },
    {
      "Id": "A.8.1",
      "Description": "User end point devices should be protected.",
      "Name": "User End Point Devices",
      "Attributes": [
        {
          "Category": "A.8 Technological controls",
          "Objetive_ID": "A.8.1",
          "Objetive_Name": "User End Point Devices",
          "Check_Summary": "Verify that endpoint protection and monitoring are enabled."
        }
      ],
      "Checks": [
        "guardduty_is_enabled",
        "ssm_managed_compliant_patching"
      ]
    },
    {
      "Id": "A.8.24",
      "Description": "Rules for the effective use of cryptography, including cryptographic key management, should be defined and implemented.",
      "Name": "Use of Cryptography",
      "Attributes": [
        {
          "Category": "A.8 Technological controls",
          "Objetive_ID": "A.8.24",
          "Objetive_Name": "Use of Cryptography",
          "Check_Summary": "Verify that encryption is enabled for data at rest and in transit."
        }
      ],
      "Checks": [
        "s3_bucket_default_encryption",
        "rds_instance_storage_encrypted",
        "ec2_ebs_volume_encryption"
      ]
    }
  ]
}

{
  "Framework": "MITRE-ATTACK",
  "Name": "MITRE ATT&CK compliance framework",
  "Version": "",
  "Provider": "AWS",
  "Description": "MITRE ATT&CK is a globally-accessible knowledge base of adversary tactics and techniques based on real-world observations. The ATT&CK knowledge base is used as a foundation for the development of specific threat models and methodologies in the private sector, in government, and in the cybersecurity product and service community.",
  "Requirements": [
    {
      "Name": "Exploit Public-Facing Application",
      "Id": "T1190",
      "Tactics": [
        "Initial Access"
      ],
      "SubTechniques": [],
      "Platforms": [
        "Containers",
        "IaaS",
        "Linux",
        "Network",
        "Windows",
        "macOS"
      ],
      "Description": "Adversaries may attempt to exploit a weakness in an Internet-facing host or system to initially access a network. The weakness in the system can be a software bug, a temporary glitch, or a misconfiguration.",
      "TechniqueURL": "https://attack.mitre.org/techniques/T1190/",
      "Checks": [
        "guardduty_is_enabled",
        "inspector2_is_enabled",
        "securityhub_enabled",
        "elbv2_waf_acl_attached",
        "awslambda_function_not_publicly_accessible",
        "ec2_instance_public_ip"
      ],
      "Attributes": [
        {
          "AWSService": "Amazon GuardDuty",
          "Category": "Detect",
          "Value": "Minimal",
          "Comment": "GuardDuty can detect when vulnerable publicly facing resources are leveraged to capture data not intended to be viewable."
        },
        {
          "AWSService": "AWS Web Application Firewall",
          "Category": "Protect",
          "Value": "Significant",
          "Comment": "AWS WAF protects public-facing applications against vulnerabilities including OWASP Top 10 via managed rule sets."
        },
        {
          "AWSService": "Amazon Inspector",
          "Category": "Protect",
          "Value": "Partial",
          "Comment": "Amazon Inspector can detect known vulnerabilities on various Windows and Linux endpoints."
        }
      ]
    },
    {
      "Name": "Valid Accounts",
      "Id": "T1078",
      "Tactics": [
        "Defense Evasion",
        "Persistence",
        "Privilege Escalation",
        "Initial Access"
      ],
      "SubTechniques": [
        "T1078.001",
        "T1078.002",
        "T1078.003",
        "T1078.004"
      ],
      "Platforms": [
        "Azure AD",
        "Containers",
        "Google Workspace",
        "IaaS",
        "Linux",
        "Network",
        "Office 365",
        "SaaS",
        "Windows",
        "macOS"
      ],
      "Description": "Adversaries may obtain and abuse credentials of existing accounts as a means of gaining Initial Access, Persistence, Privilege Escalation, or Defense Evasion.",
      "TechniqueURL": "https://attack.mitre.org/techniques/T1078/",
      "Checks": [
        "iam_root_mfa_enabled",
        "iam_user_mfa_enabled_console_access",
        "iam_no_root_access_key",
        "iam_rotate_access_key_90_days",
        "iam_user_accesskey_unused",
        "cloudtrail_multi_region_enabled"
      ],
      "Attributes": [
        {
          "AWSService": "AWS IAM",
          "Category": "Protect",
          "Value": "Significant",
          "Comment": "IAM MFA and access key rotation help prevent unauthorized access with valid credentials."
        },
        {
          "AWSService": "AWS CloudTrail",
          "Category": "Detect",
          "Value": "Significant",
          "Comment": "CloudTrail logs all API calls, enabling detection of unauthorized account usage."
        }
      ]
    },
    {
      "Name": "Data from Cloud Storage",
      "Id": "T1530",
      "Tactics": [
        "Collection"
      ],
      "SubTechniques": [],
      "Platforms": [
        "IaaS",
        "SaaS"
      ],
      "Description": "Adversaries may access data from improperly secured cloud storage. Many cloud service providers offer solutions for online data object storage.",
      "TechniqueURL": "https://attack.mitre.org/techniques/T1530/",
      "Checks": [
        "s3_bucket_public_access",
        "s3_bucket_policy_public_write_access",
        "s3_bucket_acl_prohibited",
        "s3_bucket_default_encryption",
        "macie_is_enabled"
      ],
      "Attributes": [
        {
          "AWSService": "Amazon S3",
          "Category": "Protect",
          "Value": "Significant",
          "Comment": "S3 bucket policies and ACLs can prevent public access to sensitive data."
        },
        {
          "AWSService": "Amazon Macie",
          "Category": "Detect",
          "Value": "Significant",
          "Comment": "Macie can detect and alert on sensitive data exposure in S3 buckets."
        }
      ]
    }
  ]
}

"""
FINOS Common Cloud Controls (CCC) YAML parser.

Reads cached upstream YAML files and emits Prowler-format requirements
(``{Id, Description, Attributes: [...], Checks: []}``). This module is
agnostic to providers, JSON output paths, framework metadata and legacy
check-mapping preservation — those are handled by ``sync_framework.py``.

Contract
--------
``parse_upstream(config: dict) -> list[dict]``
    Returns a list of Prowler-format requirement dicts with **guaranteed
    unique ids**. Foreign-prefix AR rewriting and genuine collision
    renumbering both happen inside this module — the runner treats id
    uniqueness as a contract violation, not as something to fix.

Config keys consumed
--------------------
This parser reads the following config entries (the rest of the config is
opaque to it):

- ``upstream.dir``              — directory containing the cached YAMLs
- ``parser.catalog_files``      — ordered list of YAML filenames to load
- ``parser.family_id_title``    — suffix → canonical family title (shape 2)
- ``parser.family_id_description`` — suffix → family description (shape 2)

Upstream shapes
---------------
FINOS CCC catalogs come in two shapes:

1. ``control-families: [{title, description, controls: [...]}]``
   (used by most catalogs)
2. ``controls: [{id, family: "CCC.X.Y", ...}]`` (no families wrapper; used
   by ``storage/object``). The ``family`` field references a family id with
   no human-readable title in the file — the title/description come from
   ``config.parser.family_id_title`` / ``family_id_description``.

Id rewriting rules
------------------
- **Foreign-prefix rewriting**: upstream intentionally aliases requirements
  across catalogs by keeping the original prefix (e.g. ``CCC.AuditLog.CN08.AR01``
  appears nested under ``CCC.Logging.CN03``). Prowler requires unique ids
  within a catalog file, so we rename the AR to fit its parent control:
  ``CCC.Logging.CN03.AR01``. See ``rewrite_ar_id()``.
- **Genuine collision renumbering**: sometimes upstream has a real typo
  where two distinct requirements share the same id (e.g.
  ``CCC.Core.CN14.AR02`` appears twice for 30-day and 14-day backup variants).
  The second copy is renumbered to the next free AR number within the
  control. See the ``seen_ids`` logic in ``emit_requirement()``.
"""
from __future__ import annotations

from pathlib import Path

import yaml


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def clean(value: str | None) -> str:
    """Trim and collapse internal whitespace/newlines into single spaces.

    Upstream YAML uses ``|`` block scalars that preserve newlines; Prowler
    stores descriptions as single-line text.
    """
    if not value:
        return ""
    return " ".join(value.split())


def flatten_mappings(mappings):
    """Convert upstream ``{reference-id, entries: [{reference-id, ...}]}`` to
    Prowler's ``{ReferenceId, Identifiers: [...]}``.
    """
    if not mappings:
        return []
    out = []
    for m in mappings:
        ids = []
        for entry in m.get("entries") or []:
            eid = entry.get("reference-id")
            if eid:
                ids.append(eid)
        out.append({"ReferenceId": m.get("reference-id", ""), "Identifiers": ids})
    return out


def ar_prefix(ar_id: str) -> str:
    """Return the first three dot-segments of an AR id (the parent control).

    e.g. ``CCC.Core.CN01.AR01`` -> ``CCC.Core.CN01``.
    """
    return ".".join(ar_id.split(".")[:3])


def rewrite_ar_id(parent_control_id: str, original_ar_id: str, ar_index: int) -> str:
    """If an AR's id doesn't share its parent control's prefix, rename it.

    Example
    -------
    parent ``CCC.Logging.CN03`` + AR id ``CCC.AuditLog.CN08.AR01`` with
    index 0 -> ``CCC.Logging.CN03.AR01``.
    """
    if ar_prefix(original_ar_id) == parent_control_id:
        return original_ar_id
    return f"{parent_control_id}.AR{ar_index + 1:02d}"


def emit_requirement(
    control: dict,
    family_name: str,
    family_desc: str,
    seen_ids: set[str],
    requirements: list[dict],
) -> None:
    """Translate one FINOS control + its assessment-requirements into
    Prowler-format requirement dicts and append them to ``requirements``.

    Applies foreign-prefix rewriting and genuine-collision renumbering so
    the final list is guaranteed to have unique ids.
    """
    control_id = clean(control.get("id"))
    control_title = clean(control.get("title"))
    section = f"{control_id} {control_title}".strip()
    objective = clean(control.get("objective"))
    threat_mappings = flatten_mappings(control.get("threat-mappings"))
    guideline_mappings = flatten_mappings(control.get("guideline-mappings"))
    ars = control.get("assessment-requirements") or []
    for idx, ar in enumerate(ars):
        raw_id = clean(ar.get("id"))
        if not raw_id:
            continue
        new_id = rewrite_ar_id(control_id, raw_id, idx)
        # Renumber on genuine upstream collision (find next free AR number)
        if new_id in seen_ids:
            base = ".".join(new_id.split(".")[:-1])
            n = 1
            while f"{base}.AR{n:02d}" in seen_ids:
                n += 1
            new_id = f"{base}.AR{n:02d}"
        seen_ids.add(new_id)

        requirements.append(
            {
                "Id": new_id,
                "Description": clean(ar.get("text")),
                "Attributes": [
                    {
                        "FamilyName": family_name,
                        "FamilyDescription": family_desc,
                        "Section": section,
                        "SubSection": "",
                        "SubSectionObjective": objective,
                        "Applicability": list(ar.get("applicability") or []),
                        "Recommendation": clean(ar.get("recommendation")),
                        "SectionThreatMappings": threat_mappings,
                        "SectionGuidelineMappings": guideline_mappings,
                    }
                ],
                "Checks": [],
            }
        )


# ---------------------------------------------------------------------------
# Public entry point
# ---------------------------------------------------------------------------


def parse_upstream(config: dict) -> list[dict]:
    """Walk upstream YAMLs and emit Prowler-format requirements.

    Handles both top-level shapes (``control-families`` and ``controls``).
    Ids are guaranteed unique in the returned list.
    """
    upstream_dir = Path(config["upstream"]["dir"])
    parser_cfg = config.get("parser") or {}
    catalog_files = parser_cfg.get("catalog_files") or []
    family_id_title = parser_cfg.get("family_id_title") or {}
    family_id_description = parser_cfg.get("family_id_description") or {}

    requirements: list[dict] = []
    seen_ids: set[str] = set()

    for filename in catalog_files:
        path = upstream_dir / filename
        if not path.exists():
            # parser.catalog_files is the closed set of upstream catalogs
            # that define the framework. Silently skipping a missing file
            # would emit valid-looking JSON with part of the framework
            # dropped, defeating the whole point of a canonical sync.
            raise FileNotFoundError(
                f"upstream catalog file not found: {path}\n"
                f"  hint: refresh the upstream cache (see SKILL.md Workflow A "
                f"Step 1), or remove {filename!r} from parser.catalog_files "
                f"if it has been retired upstream."
            )
        with open(path) as f:
            doc = yaml.safe_load(f) or {}

        # Shape 1: control-families wrapper
        for family in doc.get("control-families") or []:
            family_name = clean(family.get("title"))
            family_desc = clean(family.get("description"))
            for control in family.get("controls") or []:
                emit_requirement(
                    control, family_name, family_desc, seen_ids, requirements
                )

        # Shape 2: top-level controls with family reference id
        for control in doc.get("controls") or []:
            family_ref = clean(control.get("family"))
            suffix = family_ref.split(".")[-1] if family_ref else ""
            family_name = family_id_title.get(suffix, suffix or "Data")
            family_desc = family_id_description.get(suffix, "")
            emit_requirement(
                control, family_name, family_desc, seen_ids, requirements
            )

    return requirements

{
  "Framework": "ProwlerThreatScore",
  "Name": "Prowler ThreatScore Compliance Framework for AWS",
  "Version": "1.0",
  "Provider": "AWS",
  "Description": "Prowler ThreatScore Compliance Framework for AWS ensures that the AWS account is compliant taking into account four main pillars: Identity and Access Management, Attack Surface, Logging and Monitoring, and Encryption. Each check has a LevelOfRisk (1-5) and Weight that contribute to calculating the overall threat score.",
  "Requirements": [
    {
      "Id": "1.1.1",
      "Description": "Ensure MFA is enabled for the 'root' user account",
      "Checks": [
        "iam_root_mfa_enabled"
      ],
      "Attributes": [
        {
          "Title": "MFA enabled for 'root'",
          "Section": "1. IAM",
          "SubSection": "1.1 Authentication",
          "AttributeDescription": "The root user account holds the highest level of privileges within an AWS account. Enabling Multi-Factor Authentication (MFA) enhances security by adding an additional layer of protection beyond just a username and password.",
          "AdditionalInformation": "Enabling MFA enhances console security by requiring the authenticating user to both possess a time-sensitive key-generating device and have knowledge of their credentials.",
          "LevelOfRisk": 5,
          "Weight": 1000
        }
      ]
    },
    {
      "Id": "1.1.2",
      "Description": "Ensure hardware MFA is enabled for the 'root' user account",
      "Checks": [
        "iam_root_hardware_mfa_enabled"
      ],
      "Attributes": [
        {
          "Title": "Hardware MFA enabled for 'root'",
          "Section": "1. IAM",
          "SubSection": "1.1 Authentication",
          "AttributeDescription": "The root user account in AWS has the highest level of privileges. A hardware MFA has a smaller attack surface compared to a virtual MFA.",
          "AdditionalInformation": "Unlike a virtual MFA, which relies on a mobile device that may be vulnerable to malware, a hardware MFA operates independently, reducing exposure to potential security threats.",
          "LevelOfRisk": 5,
          "Weight": 1000
        }
      ]
    },
    {
      "Id": "1.1.13",
      "Description": "Ensure no root account access key exists",
      "Checks": [
        "iam_no_root_access_key"
      ],
      "Attributes": [
        {
          "Title": "No root access key",
          "Section": "1. IAM",
          "SubSection": "1.1 Authentication",
          "AttributeDescription": "The root account in AWS has unrestricted administrative privileges. It is recommended that no access keys be associated with the root account.",
          "AdditionalInformation": "Eliminating root access keys reduces the risk of unauthorized access and enforces the use of role-based IAM accounts with least privilege.",
          "LevelOfRisk": 5,
          "Weight": 1000
        }
      ]
    },
    {
      "Id": "2.1.1",
      "Description": "Ensure EC2 instances do not have public IP addresses",
      "Checks": [
        "ec2_instance_public_ip"
      ],
      "Attributes": [
        {
          "Title": "EC2 without public IP",
          "Section": "2. Attack Surface",
          "SubSection": "2.1 Network Exposure",
          "AttributeDescription": "EC2 instances with public IP addresses are directly accessible from the internet, increasing the attack surface.",
          "AdditionalInformation": "Use private subnets and NAT gateways or VPC endpoints for internet access when needed.",
          "LevelOfRisk": 4,
          "Weight": 100
        }
      ]
    },
    {
      "Id": "2.2.1",
      "Description": "Ensure S3 buckets are not publicly accessible",
      "Checks": [
        "s3_bucket_public_access"
      ],
      "Attributes": [
        {
          "Title": "S3 bucket not public",
          "Section": "2. Attack Surface",
          "SubSection": "2.2 Storage Exposure",
          "AttributeDescription": "Publicly accessible S3 buckets can lead to data breaches and unauthorized access to sensitive information.",
          "AdditionalInformation": "Enable S3 Block Public Access settings at the account and bucket level.",
          "LevelOfRisk": 5,
          "Weight": 1000
        }
      ]
    },
    {
      "Id": "3.1.1",
      "Description": "Ensure CloudTrail is enabled in all regions",
      "Checks": [
        "cloudtrail_multi_region_enabled"
      ],
      "Attributes": [
        {
          "Title": "CloudTrail multi-region enabled",
          "Section": "3. Logging and Monitoring",
          "SubSection": "3.1 Audit Logging",
          "AttributeDescription": "CloudTrail provides a record of API calls made in your AWS account. Multi-region trails ensure all activity is captured.",
          "AdditionalInformation": "Without comprehensive logging, security incidents may go undetected and forensic analysis becomes impossible.",
          "LevelOfRisk": 5,
          "Weight": 1000
        }
      ]
    },
    {
      "Id": "3.2.1",
      "Description": "Ensure GuardDuty is enabled",
      "Checks": [
        "guardduty_is_enabled"
      ],
      "Attributes": [
        {
          "Title": "GuardDuty enabled",
          "Section": "3. Logging and Monitoring",
          "SubSection": "3.2 Threat Detection",
          "AttributeDescription": "Amazon GuardDuty is a threat detection service that continuously monitors for malicious activity and unauthorized behavior.",
          "AdditionalInformation": "GuardDuty analyzes CloudTrail, VPC Flow Logs, and DNS logs to identify threats.",
          "LevelOfRisk": 4,
          "Weight": 100
        }
      ]
    },
    {
      "Id": "4.1.1",
      "Description": "Ensure S3 buckets have default encryption enabled",
      "Checks": [
        "s3_bucket_default_encryption"
      ],
      "Attributes": [
        {
          "Title": "S3 default encryption",
          "Section": "4. Encryption",
          "SubSection": "4.1 Data at Rest",
          "AttributeDescription": "Enabling default encryption on S3 buckets ensures all objects are encrypted when stored.",
          "AdditionalInformation": "Use SSE-S3, SSE-KMS, or SSE-C depending on your key management requirements.",
          "LevelOfRisk": 3,
          "Weight": 10
        }
      ]
    },
    {
      "Id": "4.1.2",
      "Description": "Ensure EBS volumes are encrypted",
      "Checks": [
        "ec2_ebs_volume_encryption"
      ],
      "Attributes": [
        {
          "Title": "EBS volume encryption",
          "Section": "4. Encryption",
          "SubSection": "4.1 Data at Rest",
          "AttributeDescription": "EBS volume encryption protects data at rest on EC2 instance storage.",
          "AdditionalInformation": "Enable default EBS encryption at the account level to ensure all new volumes are encrypted.",
          "LevelOfRisk": 3,
          "Weight": 10
        }
      ]
    },
    {
      "Id": "4.2.1",
      "Description": "Ensure data in transit is encrypted using TLS",
      "Checks": [
        "s3_bucket_secure_transport_policy"
      ],
      "Attributes": [
        {
          "Title": "S3 secure transport",
          "Section": "4. Encryption",
          "SubSection": "4.2 Data in Transit",
          "AttributeDescription": "Requiring HTTPS for S3 bucket access ensures data is encrypted during transmission.",
          "AdditionalInformation": "Use bucket policies to deny requests that do not use TLS.",
          "LevelOfRisk": 3,
          "Weight": 10
        }
      ]
    }
  ]
}

#!/usr/bin/env python3
"""
Keyword/service/id lookup over a Prowler check inventory produced by
build_inventory.py.

Usage:
    # Keyword AND-search across id + title + risk + description
    python skills/prowler-compliance/assets/query_checks.py aws encryption transit

    # Show all checks for a service
    python skills/prowler-compliance/assets/query_checks.py aws --service iam

    # Show full metadata for one check id
    python skills/prowler-compliance/assets/query_checks.py aws --id kms_cmk_rotation_enabled
"""
from __future__ import annotations

import json
import sys


def main() -> int:
    if len(sys.argv) < 3:
        print(__doc__)
        return 1

    provider = sys.argv[1]
    try:
        with open(f"/tmp/checks_{provider}.json") as f:
            inv = json.load(f)
    except FileNotFoundError:
        print(
            f"No inventory for {provider}. Run build_inventory.py first.",
            file=sys.stderr,
        )
        return 2

    if sys.argv[2] == "--service":
        if len(sys.argv) < 4:
            print("usage: --service <service_name>")
            return 1
        svc = sys.argv[3]
        hits = [cid for cid in sorted(inv) if inv[cid].get("service") == svc]
        for cid in hits:
            print(f"  {cid}")
            print(f"    {inv[cid].get('title', '')}")
        print(f"\n{len(hits)} checks in service '{svc}'")
    elif sys.argv[2] == "--id":
        if len(sys.argv) < 4:
            print("usage: --id <check_id>")
            return 1
        cid = sys.argv[3]
        if cid not in inv:
            print(f"NOT FOUND: {cid}")
            return 3
        m = inv[cid]
        print(f"== {cid} ==")
        print(f"service : {m.get('service')}")
        print(f"severity: {m.get('severity')}")
        print(f"resource: {m.get('resource')}")
        print(f"title   : {m.get('title')}")
        print(f"desc    : {m.get('description', '')[:500]}")
        print(f"risk    : {m.get('risk', '')[:500]}")
    else:
        keywords = [k.lower() for k in sys.argv[2:]]
        hits = 0
        for cid in sorted(inv):
            m = inv[cid]
            blob = " ".join(
                [
                    cid,
                    m.get("title", ""),
                    m.get("risk", ""),
                    m.get("description", ""),
                ]
            ).lower()
            if all(k in blob for k in keywords):
                hits += 1
                print(f"  {cid}  [{m.get('service', '')}]")
                print(f"    {m.get('title', '')[:120]}")
        print(f"\n{hits} matches for {' + '.join(keywords)}")
    return 0


if __name__ == "__main__":
    sys.exit(main())