cre.py

from cre_logging import get_logger

logger = get_logger(__name__)

import argparse
import os
import sys
import unittest
from typing import List

# NEW: load .env automatically for local development
try:
    from dotenv import load_dotenv  # type: ignore

    load_dotenv()
except ImportError:
    pass

import click  # type: ignore
from flask_migrate import Migrate  # type: ignore

from application import create_app, sqla  # type: ignore

# Hacky solutions to make this both a command line application with argparse and a flask application

app = create_app(mode=os.getenv("FLASK_CONFIG") or "default")
migrate = Migrate(app, sqla, render_as_batch=True)


# flask <x> commands
@app.cli.command()  # type: ignore
@click.option(
    "--cover/--no-cover", default=False, help="Run tests under code coverage."
)  # type: ignore
@click.argument("test_names", nargs=-1)  # type: ignore
def test(cover: bool, test_names: List[str]) -> None:
    COV = None
    if cover or os.environ.get("FLASK_COVERAGE"):
        import coverage  # type: ignore  # lazy: not needed for web/worker boot

        COV = coverage.coverage(
            branch=True,
            include="application/*",
            check_preimported=True,
            config_file="application/tests/.coveragerc",
        )
        COV.start()
        # Hack to get coverage to cover method and class defs
        from application import create_app, sqla  # type: ignore
        from application.cmd import cre_main

    if test_names:
        tests = unittest.TestLoader().loadTestsFromNames(test_names)
    else:
        tests = unittest.TestLoader().discover("application/tests", pattern="*_test.py")
    unittest.TextTestRunner(verbosity=2).run(tests)
    if COV:
        COV.stop()
        COV.save()
        print("Coverage Summary:")
        COV.report()
        basedir = os.path.abspath(os.path.dirname(__file__))
        covdir = os.path.join(basedir, "tmp/coverage")
        COV.html_report(directory=covdir)
        print("HTML version: file://%s/index.html" % covdir)
        COV.erase()


# python cre.py --<x> commands


def main() -> None:
    app_context = app.app_context()
    app_context.push()

    script_path = os.path.dirname(os.path.realpath(__file__))
    parser = argparse.ArgumentParser(
        description="Add documents describing standards to a database"
    )
    parser.add_argument(
        "--add",
        action="store_true",
        help="will treat the incoming spreadsheet as a reviewed cre and add to the database",
    )
    # parser.add_argument(
    #     "--review",
    #     action="store_true",
    #     help="will treat the incoming spreadsheet as a new mapping, will try to map the incoming connections to existing cre\
    #         and will create a new spreadsheet with the result for review. Nothing will be added to the database at this point",
    # )
    parser.add_argument(
        "--email",
        help="used in conjuctions with --review, what email to share the resulting spreadsheet with",
        default="standards_cache.sqlite",
    )
    parser.add_argument(
        "--from_spreadsheet", help="import from a spreadsheet to yaml and then database"
    )
    parser.add_argument(
        "--from_ai_exchange_csv",
        default=None,
        help="import from OpenCRE AI exchange CSV (CRE tree + MITRE ATLAS + OWASP AI Exchange)",
    )
    parser.add_argument(
        "--print_graph",
        help="will show the graph of the relationships between standards",
    )
    parser.add_argument(
        "--cache_file",
        help="where to read/store data",
        default=os.path.join(script_path, "standards_cache.sqlite"),
    )
    parser.add_argument(
        "--owasp_proj_meta",
        default=None,
        help="define location of owasp project metadata",
    )
    parser.add_argument(
        "--osib_in",
        default=None,
        help="define location of local osib file for review/add",
    )
    parser.add_argument(
        "--osib_out",
        default=None,
        help="define location of local directory to export database in OSIB format to",
    )
    # Start External Project importing
    parser.add_argument(
        "--zap_in",
        action="store_true",
        help="import zap alerts by cloning zap's website and parsing the alert .md files",
    )
    parser.add_argument(
        "--cheatsheets_in",
        action="store_true",
        help="import cheatsheets by cloning the repo website and parsing the .md files",
    )
    parser.add_argument(
        "--github_tools_in",
        action="store_true",
        help="import supported github tools, urls can be found in misc_tools_parser.py",
    )
    parser.add_argument(
        "--capec_in",
        action="store_true",
        help="import CAPEC",
    )
    parser.add_argument(
        "--cwe_in",
        action="store_true",
        help="import CWE",
    )

    parser.add_argument(
        "--csa_ccm_v3_in",
        action="store_true",
        help="import CSA's CCM v3 from https://docs.google.com/spreadsheets/d/1b5i8OV919aiqW2KcYWOQvkLorL1bRPqjthJxLH0QpD8",
    )
    parser.add_argument(
        "--csa_ccm_v4_in",
        action="store_true",
        help="import CSA's CCM v4 from https://docs.google.com/spreadsheets/d/1QDzQy0wt1blGjehyXS3uaHh7k5OOR12AWgAA1DeACyc",
    )
    parser.add_argument(
        "--iso_27001_in",
        action="store_true",
        help="import ISO 27001 by using the NIST mappings located at https://csrc.nist.gov/CSRC/media/Publications/sp/800-53/rev-5/final/documents/sp800-53r5-to-iso-27001-mapping.docx",
    )
    parser.add_argument(
        "--owasp_secure_headers_in",
        action="store_true",
        help="import owasp secure headers",
    )
    parser.add_argument(
        "--owasp_kubernetes_top10_2022_in",
        action="store_true",
        help="load OWASP Kubernetes Top Ten 2022 mapping fixture (eval gold, not import-all)",
    )
    parser.add_argument(
        "--owasp_kubernetes_top10_2025_in",
        action="store_true",
        help="load OWASP Kubernetes Top Ten 2025 mapping fixture (eval gold, not import-all)",
    )
    parser.add_argument(
        "--owasp_top10_2025_in",
        action="store_true",
        help="load OWASP Top 10 2025 mapping fixture (eval gold, not import-all)",
    )
    parser.add_argument(
        "--owasp_api_top10_2023_in",
        action="store_true",
        help="load OWASP API Security Top 10 2023 mapping fixture (eval gold, not import-all)",
    )
    parser.add_argument(
        "--owasp_llm_top10_2025_in",
        action="store_true",
        help="load OWASP LLM Top 10 2025 mapping fixture (eval gold, not import-all)",
    )
    parser.add_argument(
        "--owasp_aisvs_in",
        action="store_true",
        help="load OWASP AISVS mapping fixture (eval gold, not import-all)",
    )
    parser.add_argument(
        "--pci_dss_3_2_in",
        action="store_true",
        help="import pci dss from https://www.compliancequickstart.com/",
    )
    parser.add_argument(
        "--pci_dss_4_in",
        action="store_true",
        help="import pci dss from https://www.compliancequickstart.com/",
    )
    parser.add_argument(
        "--juiceshop_in",
        action="store_true",
        help="import juiceshop challenges from their repo",
    )
    parser.add_argument(
        "--dsomm_in",
        action="store_true",
        help="import dsomm from their repo (https://raw.githubusercontent.com/devsecopsmaturitymodel/DevSecOps-MaturityModel-data/main/src/assets/YAML/generated/generated.yaml)",
    )
    parser.add_argument(
        "--cloud_native_security_controls_in",
        action="store_true",
        help="import cloud native security controls from their repo (https://raw.githubusercontent.com/cloud-native-security-controls/controls-catalog/main/controls/controls_catalog.csv)",
    )
    # End External Project importing

    parser.add_argument(
        "--import_external_projects",
        action="store_true",
        help="import all external projects, shortcut for calling all of *_in",
    )
    parser.add_argument(
        "--generate_embeddings",
        action="store_true",
        help="for every node missing an embedding row, download content and generate embeddings",
    )
    parser.add_argument(
        "--regenerate_embeddings",
        action="store_true",
        help="delete all embedding rows then rebuild embeddings for every CRE and node (use after smart-extract or model changes)",
    )
    parser.add_argument(
        "--run_librarian",
        action="store_true",
        help="run Module C (the Librarian). With --run_id, drains Module B's "
        "knowledge_queue for that run through C.0-C.4 and marks the rows "
        "consumed; without it, walks a JSONL fixture and logs candidates",
    )
    parser.add_argument(
        "--librarian_dry_run",
        action="store_true",
        help="run the Librarian without persisting anything: no links (never "
        "written pre-W8b) and no consumed_at stamp on the queue",
    )
    parser.add_argument(
        "--librarian_source",
        default=None,
        help="path to a knowledge_queue JSONL for --run_librarian "
        "(defaults to the bundled sample fixture; not valid with --run_id)",
    )
    parser.add_argument(
        "--librarian_envelopes_out",
        default=None,
        help="Optional JSONL mirror of the Librarian's LinkProposal / ReviewItem "
        "envelopes. A real --run_id run always writes decision_queue, the durable "
        "handoff; this flag only adds a file copy of the same batch for eyeballing",
    )
    parser.add_argument(
        "--populate_neo4j_db",
        action="store_true",
        help="populate the neo4j db",
    )
    parser.add_argument(
        "--start_worker",
        action="store_true",
        help="start redis queue worker",
    )
    parser.add_argument(
        "--preload_map_analysis_target_url",
        default="",
        help="preload map analysis for all possible 2 standards combinations, use target url as an OpenCRE base",
    )
    parser.add_argument(
        "--ga_backfill_opencre_direct",
        action="store_true",
        help="refresh OpenCRE map-analysis cache rows (manual + automatic CRE links)",
    )
    parser.add_argument(
        "--ga_backfill_missing",
        action="store_true",
        help="calculate only missing directed gap-analysis pairs in the current cache DB",
    )
    parser.add_argument(
        "--ga_backfill_batch_size",
        type=int,
        default=200,
        help="batch size for queue/sync GA backfill processing",
    )
    parser.add_argument(
        "--ga_backfill_poll_seconds",
        type=int,
        default=5,
        help="poll interval (seconds) for GA backfill progress logs",
    )
    parser.add_argument(
        "--ga_backfill_max_pairs",
        type=int,
        default=0,
        help="optional cap on number of directed missing GA pairs to process (0 = no cap)",
    )
    parser.add_argument(
        "--ga_backfill_no_queue",
        action="store_true",
        help="compute missing GA pairs synchronously without RQ queue",
    )
    parser.add_argument(
        "--delete_map_analysis_for",
        default="",
        help="delete all map analyses for resource spcified",
    )
    parser.add_argument(
        "--delete_resource",
        default="",
        help="delete all the mappings, embeddings and gap analysis for the resource",
    )
    parser.add_argument(
        "--upstream_sync",
        action="store_true",
        help="download the cre graph from upstream",
    )
    parser.add_argument(
        "--export",
        action="store_true",
        help="export CRE + standards taxonomy to CSV (CI-friendly)",
    )
    parser.add_argument(
        "--csv",
        default="",
        help="output CSV path for --export",
    )
    parser.add_argument(
        "--run_noise_filter",
        action="store_true",
        help="run Module B noise/relevance filter over a harvest run's chunks",
    )
    parser.add_argument(
        "--run_harvester",
        action="store_true",
        help="run Module A harvester; writes ChangeRecords into harvest_input",
    )
    parser.add_argument(
        "--run_id",
        default="",
        help="pipeline_run_id to process (required with --run_harvester / "
        "--run_noise_filter; with --run_librarian, selects the live "
        "knowledge_queue path)",
    )
    parser.add_argument(
        "--noise_filter_dry_run",
        action="store_true",
        help="classify without writing to knowledge_queue or marking rows processed",
    )
    parser.add_argument(
        "--harvester_dry_run",
        action="store_true",
        help="run Module A without writing harvest_input rows",
    )
    parser.add_argument(
        "--harvester_repos_yaml",
        default="",
        help="optional path to repos.yaml for --run_harvester",
    )

    args = parser.parse_args()
    if args.export and not args.csv:
        parser.error("--export requires --csv <path>")
    if args.run_noise_filter and not args.run_id.strip():
        parser.error("--run_noise_filter requires --run_id <pipeline_run_id>")
    if args.run_harvester and not args.run_id.strip():
        parser.error("--run_harvester requires --run_id <pipeline_run_id>")
    # The live queue path takes its rows from the DB, so a fixture path would be
    # silently ignored rather than doing what the caller plainly asked for.
    if args.librarian_source and args.run_id.strip():
        parser.error(
            "--librarian_source reads a fixture and cannot be combined "
            "with --run_id (which drains the live knowledge_queue)"
        )

    from application.cmd import cre_main

    cre_main.run(args)


if __name__ == "__main__":  # if we're called directly
    main()