diff --git a/.gitignore b/.gitignore index ebbf2ace8..869b6f89a 100755 --- a/.gitignore +++ b/.gitignore @@ -67,6 +67,7 @@ kubescape-macos-latest *.tar.gz *.zip *_SBOM.json +*_LICENSE.json syft gitleaks_aux_report_*.json gitleaks_report.json diff --git a/docs/Docusaurus/docs/life_cycle/getting_started.md b/docs/Docusaurus/docs/life_cycle/getting_started.md index 6026d80e6..8fb7b6e66 100644 --- a/docs/Docusaurus/docs/life_cycle/getting_started.md +++ b/docs/Docusaurus/docs/life_cycle/getting_started.md @@ -34,7 +34,7 @@ For more information about structure remote config visit [Structure Remote Confi Free - ENGINE_IAC + ENGINE_IAC CHECKOV Free @@ -81,6 +81,11 @@ For more information about structure remote config visit [Structure Remote Confi TRIVY Free + + + ENGINE_LICENSE + CDXGEN + Free ENGINE_FUNCTION @@ -101,7 +106,7 @@ For more information about structure remote config visit [Structure Remote Confi ### Scan running - (CLI) - Flags ```bash -devsecops-engine-tools --platform_devops ["local","azure","github"] --remote_config_source ["local","azure","github"] --remote_config_repo ["remote_config_repo"] --remote_config_branch ["remote_config_branch"] --module ["engine_iac", "engine_dast", "engine_secret", "engine_dependencies", "engine_container", "engine_risk", "engine_code", "engine_function"] --tool ["nuclei", "bearer", "checkov", "kics", "kubescape", "trufflehog", "gitleaks", "prisma", "trivy", "xray", "dependency_check"] --folder_path ["Folder path scan engine_iac, engine_code, engine_dependencies and engine_secret"] --platform ["k8s","cloudformation","docker", "openapi", "terraform"] --use_secrets_manager ["false", "true"] --use_vulnerability_management ["false", "true"] --send_metrics ["false", "true"] --token_cmdb ["token_cmdb"] --token_vulnerability_management ["token_vulnerability_management"] --token_engine_container ["token_engine_container"] --token_engine_dependencies ["token_engine_dependencies"] --token_external_checks ["token_external_checks"] --xray_mode ["scan", "audit","build-scan"] --image_to_scan ["image_to_scan"] --dast_file_path ["dast_file_path"] --context ["false", "true"] --terraform_repo_root ["terraform_files_repo"] --docker_address ["docker addres to engine_container with Prisma tool"] +devsecops-engine-tools --platform_devops ["local","azure","github"] --remote_config_source ["local","azure","github"] --remote_config_repo ["remote_config_repo"] --remote_config_branch ["remote_config_branch"] --module ["engine_iac", "engine_dast", "engine_secret", "engine_dependencies", "engine_license", "engine_container", "engine_risk", "engine_code", "engine_function"] --tool ["nuclei", "bearer", "checkov", "kics", "kubescape", "trufflehog", "gitleaks", "prisma", "trivy", "xray", "dependency_check", "cdxgen"] --folder_path ["Folder path scan engine_iac, engine_code, engine_dependencies, engine_secret and engine_license"] --platform ["k8s","cloudformation","docker", "openapi", "terraform"] --use_secrets_manager ["false", "true"] --use_vulnerability_management ["false", "true"] --send_metrics ["false", "true"] --token_cmdb ["token_cmdb"] --token_vulnerability_management ["token_vulnerability_management"] --token_engine_container ["token_engine_container"] --token_engine_dependencies ["token_engine_dependencies"] --token_external_checks ["token_external_checks"] --xray_mode ["scan", "audit","build-scan"] --image_to_scan ["image_to_scan"] --dast_file_path ["dast_file_path"] --context ["false", "true"] --terraform_repo_root ["terraform_files_repo"] --docker_address ["docker addres to engine_container with Prisma tool"] ``` ### Scan running sample (CLI) - Local diff --git a/docs/Docusaurus/docs/life_cycle/remote_config_structure.md b/docs/Docusaurus/docs/life_cycle/remote_config_structure.md index 93461872c..84c257e46 100644 --- a/docs/Docusaurus/docs/life_cycle/remote_config_structure.md +++ b/docs/Docusaurus/docs/life_cycle/remote_config_structure.md @@ -41,6 +41,8 @@ sidebar_position: 2 ┃ ┗ 📂engine_function ┃ ┗ 📜ConfigTool.json ┃ ┗ 📜Exclusions.json + ┃ ┗ 📂engine_license + ┃ ┗ 📜ConfigTool.json ``` ## **engine_core** @@ -104,6 +106,12 @@ Software Composition Analysis (SCA) is a process that detects compromised or vul For more information about structure remote config for this module, [engine dependencies](../modules/engine_sca/engine_dependencies.md) +### **engine_license** + +Software license compliance scanning. Detects denied / unlicensed dependencies in source folders, container images, and SBOMs, and enforces a configurable policy. + +For more information about structure remote config for this module, [engine license](../modules/engine_sca/engine_license.md) + ## **vulnerability_management** ### **vulnerability_management/cmdb_mapping** diff --git a/docs/Docusaurus/docs/modules/engine_core.md b/docs/Docusaurus/docs/modules/engine_core.md index a6dd7cefd..c03e3f447 100644 --- a/docs/Docusaurus/docs/modules/engine_core.md +++ b/docs/Docusaurus/docs/modules/engine_core.md @@ -234,6 +234,10 @@ Configuration of the driven adapters in the main layer and management of on/off "TOOL": "XRAY|DEPENDENCY_CHECK|TRIVY", "PRIORITY": "STANDARD|DISCREET" }, + "ENGINE_LICENSE": { + "ENABLED": true, + "PRIORITY": "STANDARD|DISCREET" + }, "ENGINE_CODE": { "ENABLED": true, "TOOL": "BEARER|KIUWAN", @@ -440,6 +444,10 @@ Configuration of the driven adapters in the main layer and management of on/off - TOOL: XRAY | DEPENDENCY_CHECK | TRIVY - PRIORITY: STANDARD | DISCREET +- **ENGINE_LICENSE**: Configuration for the engine_license tool + - ENABLED: true or false + - PRIORITY: STANDARD | DISCREET + - **ENGINE_FUNCTION**: Configuration for the engine_function tool - ENABLED: true or false - TOOL: PRISMA diff --git a/docs/Docusaurus/docs/modules/engine_sca/engine_license.md b/docs/Docusaurus/docs/modules/engine_sca/engine_license.md new file mode 100644 index 000000000..da6013dae --- /dev/null +++ b/docs/Docusaurus/docs/modules/engine_sca/engine_license.md @@ -0,0 +1,184 @@ +# Module Engine License + +## Overview + +The `engine_license` module is a **standalone** license-compliance reporter inside the DevSecOps Engine Tools platform. It scans the local repository, generates a fresh CycloneDX SBOM via **CdxGen**, classifies every dependency against a policy declared in remote configuration, and emits a single artifact: `{pipeline_name}_LICENSE.json`. + +Unlike other engines, `engine_license` does **not**: + +- Participate in `THRESHOLD` / break-build decisions. +- Use `Exclusions.json`. +- Push findings to vulnerability management. +- Reuse SBOMs from previous runs. +- Scan container images (the artifact is repository-only). + +The intent is to ship an audit-friendly, policy-driven JSON that downstream consumers can analyse out-of-band of build pipelines. + +> **Why CdxGen?** The module relies on CdxGen because it is the only mainstream SBOM generator that enriches license data by querying the package registry (e.g. Maven Central, npm, NuGet) via its `FETCH_LICENSE` option. Tools such as Syft and Trivy report only locally declared licenses (from lockfiles or existing package metadata) and do not perform active registry lookups, so licenses missing from local manifests (common in Maven/Gradle projects) would not be resolved. This is why `SBOM_MANAGER.CDXGEN.FETCH_LICENSE` must be `true` for the report to be complete. + +## Configuration Structure + +Only one configuration file is consumed: `engine_sca/engine_license/ConfigTool.json`. There is no `Exclusions.json`; the policy is declared inline. + +### ConfigTool.json + +```json +{ + "LICENSE": { + "LICENSE_POLICY": { + "fail": ["AGPL-*", "SSPL-*"], + "warn": ["BUSL-*", "EPL-*", "LGPL-3.0*"], + "synonyms": {}, + "unlicensed_action": "ignore", + "unknown_action": "ignore" + } + } +} +``` + +#### `LICENSE_POLICY` block + +Declarative policy applied to every dependency in the SBOM. **All keys are required** for `engine_license` to produce a report; if `LICENSE_POLICY` is missing, no report is generated. + +- **fail**: Glob patterns (case-insensitive `fnmatch`) whose match marks the package as `fail` (severity `critical`). +- **warn**: Glob patterns whose match marks the package as `warn` (severity `medium`). +- **synonyms**: Object that rewrites raw license identifiers (e.g. `{"BSD": "BSD-3-Clause"}`) before matching. +- **unlicensed_action**: Action assigned to packages with no detected license. One of `fail` | `warn` | `info` | `ignore`. +- **unknown_action**: Action assigned to packages whose license label does not look like a valid SPDX identifier. Same value set as above. + +## Output Artifact: `{pipeline_name}_LICENSE.json` + +The report is written to the current working directory with a `metadata` block plus a flat `dependencies` array. + +```json +{ + "metadata": { + "pipeline_name": "my_service", + "scan_date": "2026-06-12T14:30:00", + "tool": "CDXGEN", + "policy_used": { + "fail": ["AGPL-*", "SSPL-*"], + "warn": ["BUSL-*", "EPL-*", "LGPL-3.0*"], + "synonyms": {}, + "unlicensed_action": "ignore", + "unknown_action": "ignore" + }, + "summary": { + "total_dependencies": 50, + "ok": 45, + "fail": 1, + "warn": 2, + "unlicensed": 1, + "unknown": 1 + } + }, + "dependencies": [ + { + "name": "lodash", + "version": "4.17.21", + "licenses": ["MIT"], + "policy_applied": "ok", + "policy_reason": "compliant SPDX license", + "policy_pattern_matched": null, + "license_matched": "MIT" + } + ] +} +``` + +### Field reference + +#### `metadata` +- **pipeline_name**: Value of the `pipeline_name` DevOps platform variable. +- **scan_date**: ISO-8601 timestamp when the report was assembled. +- **tool**: `"CDXGEN"`. +- **policy_used**: Deep copy of `LICENSE.LICENSE_POLICY` for auditability. +- **summary.total_dependencies**: Number of entries in `dependencies`. +- **summary.{ok,fail,warn,unlicensed,unknown}**: Count per `policy_applied` bucket. + +#### `dependencies[]` +- **name / version**: Package identifiers from the CycloneDX SBOM. +- **licenses**: List of normalized license identifiers (after applying `synonyms`). +- **policy_applied**: Bucket — one of `ok`, `fail`, `warn`, `unlicensed`, `unknown`. +- **policy_reason**: Human-readable explanation (e.g. `matches FAIL pattern 'AGPL-*'`). +- **policy_pattern_matched**: Original policy pattern that matched, or `null`. +- **license_matched**: The specific license that triggered the classification. + +### Classification rules + +For each component in the SBOM: + +1. If the component has **no licenses** → bucket `unlicensed`. +2. Otherwise the licenses are normalized through `synonyms` and each is classified: + - Match against `fail` patterns → `fail`. + - Match against `warn` patterns → `warn`. + - Looks like an SPDX id → `ok`. + - Otherwise → `unknown`. +3. **Highest-risk-wins:** if any license on the package matches `fail`, the package is `fail`; if any matches `warn`, it's `warn`. Only when all licenses are compliant is the package `ok`. + +## Console Output (Findings) + +When run with `--context false` (default), `engine_license` prints a compliance table showing only `fail` and `warn` findings: + +| Severity | ID | Description | Where | +|---|---|---|---| +| critical | AGPL-3.0-itext | License 'AGPL-3.0' for package 'itext' (...) | itext:9.2.0 | +| medium | EPL-2.0-junit-bom | License 'EPL-2.0' for package 'junit-bom' (...) | junit-bom:5.10.2 | + +Dependencies classified as `ok`, `unlicensed`, or `unknown` do **not** appear in the console findings. + +## Structured Context (`--context true`) + +When run with `--context true`, the engine additionally prints a JSON block: + +``` +===== BEGIN CONTEXT OUTPUT ===== +{ + "license_context": [ + { + "name": "itext", + "version": "9.2.0", + "licenses": ["AGPL-3.0"], + "policy_applied": "fail", + "policy_reason": "matches FAIL pattern 'AGPL-*'", + "policy_pattern_matched": "AGPL-*", + "severity": "critical", + "priority": "very critical" + } + ] +} +===== END CONTEXT OUTPUT ===== +``` + +Only `fail` and `warn` dependencies appear in the context output. + +## Key Components + +- `applications/runner_license_scan.py`: Entry point invoked by `engine_core/handle_scan`. Runs the flow and builds `Finding` objects for the compliance table. +- `infrastructure/entry_points/entry_point_tool.py`: Orchestrator: fetch remote config → fresh SBOM → build report. +- `infrastructure/driven_adapters/license_scan/license_scan_manager.py`: Reads the LICENSE.json and provides structured context extraction. +- `infrastructure/helpers/license_policy.py`: Pure helpers — `build_policy_from_remote_config`, `classify_package`, `looks_like_spdx_id`. +- `domain/usecases/build_license_report.py`: `BuildLicenseReport` use case that reads the CycloneDX SBOM, classifies every dependency, and writes the LICENSE.json artifact. + +## Example Usage + +```sh +devsecops-engine-tools \ + --platform_devops local \ + --remote_config_source local \ + --remote_config_repo example_remote_config_local \ + --module engine_license \ + --folder_path path/to/project +``` + +If `--folder_path` is omitted, the current working directory is scanned. + +## Configuration Guidelines + +- Keep `LICENSE_POLICY` in remote configuration so policy changes are reviewed and auditable; the report echoes it verbatim under `metadata.policy_used`. +- Use specific SPDX identifiers in `fail` / `warn` patterns when possible (e.g. `AGPL-3.0`, `LGPL-3.0*`); use globs sparingly. +- Use `synonyms` to canonicalize ambiguous labels from the SBOM (e.g. `"BSD"` → `"BSD-3-Clause"`). +- Ensure `SBOM_MANAGER.CDXGEN.FETCH_LICENSE` is `true` so the SBOM includes license metadata. +- Choose `unlicensed_action` and `unknown_action` deliberately: + - `ignore` keeps the package in the report but assigns `info` severity (not shown in findings). + - `warn` / `fail` raises the severity for those buckets. diff --git a/example_remote_config_local/engine_core/ConfigTool.json b/example_remote_config_local/engine_core/ConfigTool.json index 21bbaab6b..2aa53b8da 100644 --- a/example_remote_config_local/engine_core/ConfigTool.json +++ b/example_remote_config_local/engine_core/ConfigTool.json @@ -217,6 +217,10 @@ "ENABLED": true, "TOOL": "XRAY|DEPENDENCY_CHECK|TRIVY" }, + "ENGINE_LICENSE": { + "ENABLED": true, + "PRIORITY": "DISCREET" + }, "ENGINE_CODE": { "ENABLED": true, "TOOL": "BEARER|KIUWAN", diff --git a/example_remote_config_local/engine_sca/engine_license/ConfigTool.json b/example_remote_config_local/engine_sca/engine_license/ConfigTool.json new file mode 100644 index 000000000..a52c4cc81 --- /dev/null +++ b/example_remote_config_local/engine_sca/engine_license/ConfigTool.json @@ -0,0 +1,24 @@ +{ + "LICENSE": { + "LICENSE_POLICY": { + "fail": ["AGPL-*", "SSPL-*"], + "warn": ["BUSL-*", "EPL-*", "LGPL-3.0*"], + "synonyms": { + }, + "unlicensed_action": "ignore", + "unknown_action": "ignore", + "severity_mapping": { + "fail": "critical", + "warn": "medium", + "ok": "info", + "ignore": "info" + } + } + }, + "THRESHOLD": { + "VULNERABILITY": {}, + "COMPLIANCE": { + "Critical": 1 + } + } +} diff --git a/tools/devsecops_engine_tools/engine_core/src/applications/runner_engine_core.py b/tools/devsecops_engine_tools/engine_core/src/applications/runner_engine_core.py index 64fa38c26..abc0c0dae 100755 --- a/tools/devsecops_engine_tools/engine_core/src/applications/runner_engine_core.py +++ b/tools/devsecops_engine_tools/engine_core/src/applications/runner_engine_core.py @@ -116,6 +116,7 @@ def get_inputs_from_cli(args): "xray", "dependency_check", "kiuwan", + "cdxgen", "all_tools", ], type=str, @@ -132,6 +133,7 @@ def get_inputs_from_cli(args): "engine_secret", "engine_dependencies", "engine_container", + "engine_license", "engine_risk", "engine_function", ], @@ -266,6 +268,7 @@ def get_inputs_from_cli(args): "engine_secret": ["trufflehog", "gitleaks", "all_tools"], "engine_container": ["prisma", "trivy"], "engine_dependencies": ["xray", "dependency_check", "trivy"], + "engine_license": ["cdxgen"], "engine_code": ["bearer", "kiuwan"], "engine_dast": ["nuclei"], "engine_risk": None, diff --git a/tools/devsecops_engine_tools/engine_core/src/domain/usecases/handle_scan.py b/tools/devsecops_engine_tools/engine_core/src/domain/usecases/handle_scan.py index 4732c3e38..a9a6bd538 100644 --- a/tools/devsecops_engine_tools/engine_core/src/domain/usecases/handle_scan.py +++ b/tools/devsecops_engine_tools/engine_core/src/domain/usecases/handle_scan.py @@ -46,6 +46,12 @@ from devsecops_engine_tools.engine_sca.engine_dependencies.src.applications.runner_dependencies_scan import ( runner_engine_dependencies, ) +from devsecops_engine_tools.engine_sca.engine_license.src.applications.runner_license_scan import ( + runner_engine_license, +) +from devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.driven_adapters.license_scan.license_scan_manager import ( + LicenseScanManager, +) from devsecops_engine_tools.engine_sca.engine_function.src.applications.runner_function_scan import ( runner_engine_function, ) @@ -220,6 +226,27 @@ def process(self, dict_args: any, config_tool: any): config_tool, input_core, dict_args, secret_tool, env, sbom_components ) + self.risk_score_gateway.get_risk_score(findings_list, config_tool, dict_args["module"]) + return findings_list, input_core + elif "engine_license" in dict_args["module"]: + findings_list, input_core, _sbom_components = runner_engine_license( + dict_args, + config_tool, + secret_tool, + self.devops_platform_gateway, + self.remote_config_source_gateway, + self.sbom_tool_gateway, + ) + + self._handle_context_extraction( + dict_args, + "engine_license", + input_core.path_file_results, + config_tool["ENGINE_LICENSE"], + LicenseScanManager(), + config_tool + ) + self.risk_score_gateway.get_risk_score(findings_list, config_tool, dict_args["module"]) return findings_list, input_core diff --git a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/context_extraction/context_extraction_manager.py b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/context_extraction/context_extraction_manager.py index 19661c8c6..e30aad995 100644 --- a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/context_extraction/context_extraction_manager.py +++ b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/context_extraction/context_extraction_manager.py @@ -25,6 +25,7 @@ def __init__(self, risk_score_gateway: RiskScoreGateway): "engine_iac": "get_iac_context_from_results", "engine_container": "get_container_context_from_results", "engine_dependencies": "get_dependencies_context_from_results", + "engine_license": "get_license_context_from_results", } # Mapping of module names to their context output keys @@ -32,6 +33,7 @@ def __init__(self, risk_score_gateway: RiskScoreGateway): "engine_iac": "iac_context", "engine_container": "container_context", "engine_dependencies": "dependencies_context", + "engine_license": "license_context", } def register_tool_gateway(self, module_name: str, tool_gateway: any): @@ -75,7 +77,7 @@ def extract_context( # Dependencies requires remote_config context_list = method(path_file_results, remote_config=remote_config, **kwargs) else: - # IaC and Container only need path_file_results + # IaC, Container, and License only need path_file_results context_list = method(path_file_results) if not context_list: @@ -116,6 +118,9 @@ def _calculate_priorities_for_context( elif hasattr(context_item, 'id'): # For IaC: id is a string finding_id = context_item.id + elif hasattr(context_item, 'name'): + # For license: name is the identifier + finding_id = context_item.name else: finding_id = "unknown" diff --git a/tools/devsecops_engine_tools/engine_core/test/domain/usecases/test_handle_scan.py b/tools/devsecops_engine_tools/engine_core/test/domain/usecases/test_handle_scan.py index 949ea1a40..d292f3a23 100644 --- a/tools/devsecops_engine_tools/engine_core/test/domain/usecases/test_handle_scan.py +++ b/tools/devsecops_engine_tools/engine_core/test/domain/usecases/test_handle_scan.py @@ -984,3 +984,91 @@ def test_define_threshold_quality_vuln_no_matching_pt(self, mock_runner_engine_c self.handle_scan.process(dict_args, config_tool) + + @mock.patch( + "devsecops_engine_tools.engine_core.src.domain.usecases.handle_scan.runner_engine_license" + ) + def test_process_with_engine_license_returns_runner_input_core( + self, mock_runner_engine_license + ): + """engine_license is standalone: runner builds the InputCore and the + handle_scan branch just forwards it. No VM, no risk_score.""" + dict_args = { + "use_secrets_manager": "false", + "module": "engine_license", + "use_vulnerability_management": "true", + "remote_config_repo": "test_repo", + "remote_config_branch": "", + } + config_tool = { + "VULNERABILITY_MANAGER": {}, + "ENGINE_LICENSE": {"ENABLED": "true"}, + } + + runner_input_core = InputCore( + totalized_exclusions=[], + threshold_defined=self.threshold, + path_file_results="/abs/svc_LICENSE.json", + custom_message_break_build="License scan completed", + scope_pipeline="svc", + scope_service="svc", + stage_pipeline="Build", + ) + mock_runner_engine_license.return_value = ( + [], + runner_input_core, + ["sbom_component"], + ) + + findings, input_core = self.handle_scan.process(dict_args, config_tool) + + self.assertEqual(findings, []) + self.assertIs(input_core, runner_input_core) + + self.vulnerability_management.send_vulnerability_management.assert_not_called() + self.risk_score_gateway.get_risk_score.assert_called_once() + + mock_runner_engine_license.assert_called_once_with( + dict_args, + config_tool, + mock.ANY, # secret_tool + self.devops_platform_gateway, + self.remote_config_source_gateway, + self.sbom_gateway, + ) + + @mock.patch( + "devsecops_engine_tools.engine_core.src.domain.usecases.handle_scan.runner_engine_license" + ) + def test_process_with_engine_license_propagates_runner_failure_input_core( + self, mock_runner_engine_license + ): + """When the runner reports a failed license report, handle_scan still + returns the InputCore the runner produced.""" + dict_args = { + "use_secrets_manager": "false", + "module": "engine_license", + "use_vulnerability_management": "true", + "remote_config_repo": "test_repo", + "remote_config_branch": "", + } + config_tool = { + "VULNERABILITY_MANAGER": {}, + "ENGINE_LICENSE": {"ENABLED": "true"}, + } + + failure_input_core = InputCore( + totalized_exclusions=[], + threshold_defined=self.threshold, + path_file_results=None, + custom_message_break_build="License scan completed", + scope_pipeline="svc", + scope_service="svc", + stage_pipeline="Build", + ) + mock_runner_engine_license.return_value = ([], failure_input_core, None) + + findings, input_core = self.handle_scan.process(dict_args, config_tool) + self.assertEqual(findings, []) + self.assertIsNone(input_core.path_file_results) + self.assertEqual(input_core.scope_pipeline, "svc") diff --git a/tools/devsecops_engine_tools/engine_core/test/infrastructure/driven_adapters/context_extraction/test_context_extraction_manager.py b/tools/devsecops_engine_tools/engine_core/test/infrastructure/driven_adapters/context_extraction/test_context_extraction_manager.py index 4dbdd8720..3bf5337b2 100644 --- a/tools/devsecops_engine_tools/engine_core/test/infrastructure/driven_adapters/context_extraction/test_context_extraction_manager.py +++ b/tools/devsecops_engine_tools/engine_core/test/infrastructure/driven_adapters/context_extraction/test_context_extraction_manager.py @@ -27,7 +27,7 @@ def test_initialization(self): self.assertIsNotNone(self.manager._tool_gateways) self.assertIsNotNone(self.manager._method_mapping) self.assertEqual(len(self.manager._tool_gateways), 0) - self.assertEqual(len(self.manager._method_mapping), 3) + self.assertEqual(len(self.manager._method_mapping), 4) self.assertIsNotNone(self.manager._risk_score_gateway) def test_register_tool_gateway_iac(self): @@ -303,5 +303,84 @@ def test_extract_context_with_gateway_missing_method(self): self.config_tool ) + def test_extract_context_license(self): + """Test extracting context for License module.""" + path_file_results = "/path/to/svc_LICENSE.json" + mock_license_gateway = Mock() + mock_license_gateway.get_license_context_from_results.return_value = [] + self.manager.register_tool_gateway("engine_license", mock_license_gateway) + + self.manager.extract_context( + "engine_license", + path_file_results, + self.remote_config, + self.config_tool + ) + + mock_license_gateway.get_license_context_from_results.assert_called_once_with( + path_file_results + ) + + def test_extract_context_license_prints_output(self): + """Test that license context extraction prints the correct JSON between markers.""" + from io import StringIO + import sys + from dataclasses import dataclass, field + from typing import List, Optional + + @dataclass + class FakeContextLicense: + name: str + version: str + licenses: List[str] + policy_applied: str + policy_reason: str + policy_pattern_matched: str + severity: str + priority: Optional[str] = field(default=None) + + context_items = [ + FakeContextLicense( + name="ngrx", + version="1.0.0", + licenses=["AGPL-3.0"], + policy_applied="fail", + policy_reason="Matched AGPL-*", + policy_pattern_matched="AGPL-*", + severity="critical", + ) + ] + + mock_license_gateway = Mock() + mock_license_gateway.get_license_context_from_results.return_value = context_items + self.manager.register_tool_gateway("engine_license", mock_license_gateway) + + # Mock risk_score_gateway to set priority + def set_priority(findings, config_tool, module): + from devsecops_engine_tools.engine_core.src.domain.model.finding import Priority + for f in findings: + f.priority = Priority(score=1.0, scale="very critical") + + self.mock_risk_score_gateway.get_risk_score.side_effect = set_priority + + captured = StringIO() + sys.stdout = captured + try: + self.manager.extract_context( + "engine_license", + "/path/to/svc_LICENSE.json", + self.remote_config, + self.config_tool + ) + finally: + sys.stdout = sys.__stdout__ + + output = captured.getvalue() + assert "===== BEGIN CONTEXT OUTPUT =====" in output + assert "===== END CONTEXT OUTPUT =====" in output + assert '"license_context"' in output + assert '"ngrx"' in output + assert '"very critical"' in output + if __name__ == '__main__': unittest.main() diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/applications/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/applications/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/applications/runner_license_scan.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/applications/runner_license_scan.py new file mode 100644 index 000000000..a7a5a775a --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/src/applications/runner_license_scan.py @@ -0,0 +1,95 @@ +import json + +from devsecops_engine_tools.engine_core.src.domain.model.finding import ( + Finding, + Category, +) +from devsecops_engine_tools.engine_core.src.domain.model.input_core import InputCore +from devsecops_engine_tools.engine_core.src.domain.model.threshold import Threshold +from devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.entry_points.entry_point_tool import ( + init_engine_license, +) + +_RELEVANT_POLICIES = ("fail", "warn") +_DEFAULT_COMPLIANCE_THRESHOLD = {"Critical": 1} + +def _build_findings_from_license_json(license_json_path): + if not license_json_path: + return [] + with open(license_json_path, "r") as fh: + data = json.load(fh) + findings = [] + for dep in data.get("dependencies", []): + policy = dep.get("policy_applied", "") + if policy not in _RELEVANT_POLICIES: + continue + name = dep.get("name", "unknown") + version = dep.get("version", "") + license_label = dep.get("license_matched") or (dep.get("licenses", []) or ["UNKNOWN"])[0] + findings.append( + Finding( + id=f"{license_label}-{name}", + cvss="", + where=f"{name}:{version}", + description=f"License '{license_label}' for package '{name}' ({dep.get('policy_reason', '')}). ", + severity=dep.get("severity", "info"), + identification_date="", + published_date_cve="", + module="engine_license", + category=Category.COMPLIANCE, + requirements="", + tool="CDXGEN", + ) + ) + return findings + + +def runner_engine_license( + dict_args, + config_tool, + secret_tool, + devops_platform_gateway, + remote_config_source_gateway, + sbom_tool_gateway, +): + """Run the engine_license standalone flow. + + Produces ''{pipeline_name}_LICENSE.json'' in the CWD. + + Returns: + Tuple ''{findings_list, input_core, sbom_components}''. + """ + try: + license_json_path, sbom_components, remote_config = init_engine_license( + devops_platform_gateway, + remote_config_source_gateway, + dict_args, + config_tool, + sbom_tool_gateway, + ) + + pipeline_name = devops_platform_gateway.get_variable("pipeline_name") + findings_list = _build_findings_from_license_json(license_json_path) + + threshold_data = (remote_config or {}).get("THRESHOLD", {}) + input_core = InputCore( + totalized_exclusions=[], + threshold_defined=Threshold({ + "VULNERABILITY": threshold_data.get("VULNERABILITY", {}), + "COMPLIANCE": threshold_data.get("COMPLIANCE", _DEFAULT_COMPLIANCE_THRESHOLD), + }), + path_file_results=license_json_path, + custom_message_break_build="License scan completed", + scope_pipeline=pipeline_name, + scope_service=pipeline_name, + stage_pipeline="Build", + ) + + return findings_list, input_core, sbom_components + + except Exception as e: + raise Exception(f"Error SCAN engine license : {str(e)}") + + +if __name__ == "__main__": + runner_engine_license() diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/model/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/model/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/model/context_license.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/model/context_license.py new file mode 100644 index 000000000..756684115 --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/model/context_license.py @@ -0,0 +1,14 @@ +from dataclasses import dataclass, field +from typing import List, Optional + + +@dataclass +class ContextLicense: + name: str + version: str + licenses: List[str] + policy_applied: str + policy_reason: str + policy_pattern_matched: str + severity: str + priority: Optional[str] = field(default=None) diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/model/gateways/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/model/gateways/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/usecases/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/usecases/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/usecases/build_license_report.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/usecases/build_license_report.py new file mode 100644 index 000000000..27c5a7266 --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/src/domain/usecases/build_license_report.py @@ -0,0 +1,138 @@ +"""Use case that builds the LICENSE.json artifact from a CycloneDX SBOM. + +The output is a hybrid JSON: top-level ``metadata`` block (pipeline name, +scan date, tool, applied policy echo, summary counts) plus a flat +``dependencies`` array listing every package found in the SBOM with its +license(s) and the policy decision. + +The artifact is written to ``{pipeline_name}_LICENSE.json`` in the chosen +output directory (CWD by default). +""" + +import copy +import json +import os +from datetime import datetime + +from devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.helpers.license_policy import ( + build_policy_from_remote_config, + classify_package, +) +from devsecops_engine_tools.engine_utilities.utils.logger_info import MyLogger +from devsecops_engine_tools.engine_utilities import settings + +logger = MyLogger.__call__(**settings.SETTING_LOGGER).get_logger() + +TOOL = "CDXGEN" +_SUMMARY_BUCKETS = ("ok", "fail", "warn", "unlicensed", "unknown") + + +class BuildLicenseReport: + """Build the LICENSE.json artifact from a CycloneDX SBOM. + + The ``process`` method returns the absolute path to the file it writes + or ``None`` when no report could be generated. + """ + + def __init__(self, output_dir: str = None): + self.output_dir = output_dir or os.getcwd() + + def process(self, sbom_path, remote_config, pipeline_name): + policy = build_policy_from_remote_config(remote_config) + if policy is None: + logger.error( + "Cannot build LICENSE report: LICENSE_POLICY missing in remote config." + ) + return None + + data = self._read_sbom(sbom_path) + if data is None: + return None + + dependencies = self._build_dependencies(data, policy) + report = { + "metadata": self._build_metadata( + pipeline_name, remote_config, dependencies + ), + "dependencies": dependencies, + } + return self._write_report(report, pipeline_name) + + @staticmethod + def _read_sbom(sbom_path): + if not sbom_path: + logger.error("SBOM path is empty; cannot build LICENSE report.") + return None + if not os.path.exists(sbom_path): + logger.error(f"SBOM not found: {sbom_path}") + return None + try: + with open(sbom_path, "r") as fh: + return json.load(fh) + except Exception as e: + logger.error(f"Error reading SBOM '{sbom_path}': {e}") + return None + + def _build_dependencies(self, data, policy): + components = data.get("components", []) + dependencies = [] + for component in components: + pkg_name = component.get("name", "unknown") + pkg_version = component.get("version", "") + raw_licenses = component.get("licenses", []) + licenses = [ + entry.get("license", entry) for entry in raw_licenses + if isinstance(entry, dict) + ] + + classification = classify_package(licenses, policy) + dependencies.append( + { + "name": pkg_name, + "version": pkg_version, + "licenses": classification["licenses"], + "policy_applied": classification["policy_applied"], + "policy_reason": classification["reason"], + "policy_pattern_matched": classification[ + "pattern_matched" + ], + "license_matched": classification["label"], + "severity": classification["severity"], + } + ) + return dependencies + + def _build_metadata(self, pipeline_name, remote_config, dependencies): + policy_used = copy.deepcopy( + (remote_config or {}).get("LICENSE", {}).get("LICENSE_POLICY", {}) + ) + + summary = { + "total_dependencies": len(dependencies), + } + for bucket in _SUMMARY_BUCKETS: + summary[bucket] = sum( + 1 for d in dependencies if d["policy_applied"] == bucket + ) + + return { + "pipeline_name": pipeline_name, + "scan_date": datetime.now().isoformat(timespec="seconds"), + "tool": TOOL, + "policy_used": policy_used, + "summary": summary, + } + + def _write_report(self, report, pipeline_name): + os.makedirs(self.output_dir, exist_ok=True) + file_name = f"{pipeline_name}_LICENSE.json" + path = os.path.join(self.output_dir, file_name) + try: + with open(path, "w") as fh: + json.dump(report, fh, indent=2) + abs_path = os.path.abspath(path) + logger.info(f"License report saved to: {abs_path}") + return abs_path + except Exception as e: + logger.error(f"Error writing LICENSE report '{path}': {e}") + return None diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/driven_adapters/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/driven_adapters/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/driven_adapters/license_scan/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/driven_adapters/license_scan/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/driven_adapters/license_scan/license_scan_manager.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/driven_adapters/license_scan/license_scan_manager.py new file mode 100644 index 000000000..16a32fd47 --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/driven_adapters/license_scan/license_scan_manager.py @@ -0,0 +1,34 @@ +import json +from typing import List + +from devsecops_engine_tools.engine_sca.engine_license.src.domain.model.context_license import ( + ContextLicense, +) + +_RELEVANT_POLICIES = ("fail", "warn") + + +class LicenseScanManager: + """Reads the LICENSE.json and provides context extraction for engine_license.""" + + def get_license_context_from_results(self, path_file_results) -> List[ContextLicense]: + with open(path_file_results, "r") as fh: + data = json.load(fh) + + context_list = [] + for dep in data.get("dependencies", []): + policy = dep.get("policy_applied", "unknown") + if policy not in _RELEVANT_POLICIES: + continue + context_list.append( + ContextLicense( + name=dep.get("name", "unknown"), + version=dep.get("version", ""), + licenses=dep.get("licenses", []), + policy_applied=policy, + policy_reason=dep.get("policy_reason", ""), + policy_pattern_matched=dep.get("policy_pattern_matched", "") or "", + severity=dep.get("severity", "info"), + ) + ) + return context_list diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/entry_points/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/entry_points/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/entry_points/entry_point_tool.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/entry_points/entry_point_tool.py new file mode 100644 index 000000000..8704e9138 --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/entry_points/entry_point_tool.py @@ -0,0 +1,71 @@ +"""Entry point of the engine_license module. + +The flow is: + + 1. Generate a fresh SBOM via CdxGen. + 2. Build the ``{pipeline_name}_LICENSE.json`` artifact directly from + the SBOM, applying the policy declared in remote_config. + +The entry point returns ``(license_json_path, sbom_components)``. +""" + +import os + +from devsecops_engine_tools.engine_sca.engine_license.src.domain.usecases.build_license_report import ( + BuildLicenseReport, +) +from devsecops_engine_tools.engine_core.src.domain.model.gateway.devops_platform_gateway import ( + DevopsPlatformGateway, +) +from devsecops_engine_tools.engine_core.src.domain.model.gateway.sbom_manager import ( + SbomManagerGateway, +) +from devsecops_engine_tools.engine_utilities.utils.logger_info import MyLogger +from devsecops_engine_tools.engine_utilities import settings + +logger = MyLogger.__call__(**settings.SETTING_LOGGER).get_logger() + + +def init_engine_license( + devops_platform_gateway: DevopsPlatformGateway, + remote_config_source_gateway, + dict_args, + config_tool, + tool_sbom: SbomManagerGateway, +): + """Run the standalone engine_license flow. + + Returns a tuple ``(license_json_path, sbom_components, remote_config)``. + ``license_json_path``/``sbom_components`` may be ``None`` if the + corresponding step failed. ``remote_config`` is always the module's + remote config (used e.g. to build the compliance threshold). + """ + remote_config = remote_config_source_gateway.get_remote_config( + dict_args["remote_config_repo"], + "engine_sca/engine_license/ConfigTool.json", + dict_args["remote_config_branch"], + ) + + pipeline_name = devops_platform_gateway.get_variable("pipeline_name") + to_scan = dict_args.get("folder_path") or os.getcwd() + + if not os.path.exists(to_scan): + logger.error(f"Path {to_scan} does not exist; aborting license scan.") + return None, None, remote_config + + config_sbom = config_tool.get("SBOM_MANAGER", {}) or {} + if tool_sbom is None: + logger.error("SBOM tool gateway is not configured; aborting license scan.") + return None, None, remote_config + sbom_components = tool_sbom.get_components(to_scan, config_sbom, pipeline_name) + sbom_path = f"{pipeline_name}_SBOM.json" + if not os.path.exists(sbom_path): + logger.error( + f"SBOM file {sbom_path} not found after generation; aborting license scan." + ) + return None, sbom_components, remote_config + + license_json_path = BuildLicenseReport().process( + sbom_path, remote_config, pipeline_name + ) + return license_json_path, sbom_components, remote_config diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/helpers/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/helpers/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/helpers/license_policy.py b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/helpers/license_policy.py new file mode 100644 index 000000000..5243aae7d --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/src/infrastructure/helpers/license_policy.py @@ -0,0 +1,238 @@ +"""Pure license classification helpers used by the engine_license module. + +Functions in this module are stateless and do not perform I/O. They are +shared by the LICENSE.json builder and any other consumer that needs to +classify SBOM packages against a remote_config policy. + +Policy buckets exposed via ``classify_package`` map 1:1 to the summary +buckets in the LICENSE.json artifact: ``ok``, ``fail``, ``warn``, +``unlicensed`` and ``unknown``. +""" + +import fnmatch +import re + +from devsecops_engine_tools.engine_utilities.utils.logger_info import MyLogger +from devsecops_engine_tools.engine_utilities import settings + +logger = MyLogger.__call__(**settings.SETTING_LOGGER).get_logger() + +TOOL = "LICENSE" + +_DEFAULT_SEVERITY_BY_ACTION = { + "fail": "critical", + "warn": "medium", + "ok": "info", + "ignore": "info", +} + +__DEFAULT_VALID_ACTIONS = {"fail", "warn", "info", "ignore"} + + +def get_value(obj, *keys, default=None): + """Return the first non-None value among the given key variants.""" + if not isinstance(obj, dict): + return default + for key in keys: + if key in obj and obj[key] is not None: + return obj[key] + return default + + +def validate_action(value, default): + """Return value (lowercased) if it is a valid action, else default.""" + v = str(value).lower().strip() + return v if v in __DEFAULT_VALID_ACTIONS else default + + +def looks_like_spdx_id(label): + """Heuristic: SPDX ids are short, no spaces, alphanumeric + a few symbols.""" + if not label or len(label) > 60: + return False + if " " in label: + return False + return bool(re.match(r"^[A-Za-z0-9.\-+]+$", label)) + + +def build_policy_from_remote_config(remote_config): + """Build the policy dict from remote_config LICENSE_POLICY section. + + Returns None (and logs error) if the configuration is absent. + """ + if not (remote_config and isinstance(remote_config, dict)): + logger.error("No remote_config provided: LICENSE_POLICY is required.") + return None + + override = (remote_config.get(TOOL) or {}).get("LICENSE_POLICY") + if not isinstance(override, dict): + logger.error( + "remote_config missing LICENSE_POLICY configuration: cannot classify licenses." + ) + return None + + fail_raw = override.get("fail", []) + warn_raw = override.get("warn", []) + + fail_list = [str(p) for p in fail_raw] if isinstance(fail_raw, list) else [] + warn_list = [str(p) for p in warn_raw] if isinstance(warn_raw, list) else [] + + synonyms_raw = override.get("synonyms", {}) + synonyms = { + str(k): str(v) + for k, v in (synonyms_raw.items() if isinstance(synonyms_raw, dict) else []) + } + + severity_mapping = dict(_DEFAULT_SEVERITY_BY_ACTION) + severity_mapping_raw = override.get("severity_mapping", {}) + if isinstance(severity_mapping_raw, dict): + severity_mapping.update( + {str(k): str(v) for k, v in severity_mapping_raw.items()} + ) + + return { + "fail": fail_list, + "warn": warn_list, + "synonyms": synonyms, + "unlicensed_action": validate_action( + override.get("unlicensed_action", "ignore"), "ignore" + ), + "unknown_action": validate_action( + override.get("unknown_action", "ignore"), "ignore" + ), + "severity_mapping": severity_mapping, + } + + +def _classify_label(normalized_label, policy): + """Classify a single normalized license label. + + Returns a dict with: bucket ∈ {fail, warn, ok, unknown}, reason, + pattern_matched (str|None). + """ + label = (normalized_label or "").strip() + label_lc = label.lower() + + for pattern in policy["fail"]: + if fnmatch.fnmatchcase(label_lc, pattern.lower()): + return { + "bucket": "fail", + "reason": f"matches FAIL pattern '{pattern}'", + "pattern_matched": pattern, + } + + for pattern in policy["warn"]: + if fnmatch.fnmatchcase(label_lc, pattern.lower()): + return { + "bucket": "warn", + "reason": f"matches WARN pattern '{pattern}'", + "pattern_matched": pattern, + } + + if looks_like_spdx_id(label): + return { + "bucket": "ok", + "reason": "compliant SPDX license", + "pattern_matched": None, + } + + return { + "bucket": "unknown", + "reason": "non-SPDX license label", + "pattern_matched": None, + } + + +def _extract_license_id(license_entry): + """Best-effort extraction of a license identifier from a CycloneDX SBOM entry.""" + return ( + get_value( + license_entry, + "id", "Id", "ID", + "spdxExpression", "SpdxExpression", + "name", "Name", + default="", + ) + or "" + ) + + +def classify_package(licenses, policy): + """Classify all licenses on a single package against the policy. + + Highest-risk-wins semantics: if ANY license on the package matches a + ``fail`` pattern, the package is classified as ``fail``. Otherwise, if + any matches ``warn``, the package is ``warn``. Only when all licenses are + compliant (or a single ok is found with no risky siblings) the package is + ``ok``. + + Args: + licenses: list of license entries (dicts) from the CycloneDX SBOM. + May be empty, in which case the package is considered unlicensed. + policy: policy dict produced by ``build_policy_from_remote_config``. + + Returns: + dict with keys: + - policy_applied: one of {ok, fail, warn, unlicensed, unknown} + - label: normalized license label that was applied (or + "UNLICENSED"/"UNKNOWN" sentinels) + - licenses: list[str] of all normalized license labels detected + - reason: human-readable explanation + - pattern_matched: pattern from policy that matched, or None + - severity: effective severity tier for this entry + """ + if policy is None: + return { + "policy_applied": "unknown", + "label": "UNKNOWN", + "licenses": [], + "reason": "no policy available", + "pattern_matched": None, + "severity": _DEFAULT_SEVERITY_BY_ACTION["ignore"], + } + + if not licenses: + action = policy["unlicensed_action"] + return { + "policy_applied": "unlicensed", + "label": "UNLICENSED", + "licenses": [], + "reason": "no license detected", + "pattern_matched": None, + "severity": policy["severity_mapping"].get(action, "info"), + } + + normalized_labels = [] + classified = [] + for lic in licenses: + raw_id = _extract_license_id(lic) + normalized = ( + policy["synonyms"].get(raw_id, raw_id).strip() or "UNKNOWN" + ) + normalized_labels.append(normalized) + classified.append((normalized, _classify_label(normalized, policy))) + + # Highest risk wins: fail > warn > unknown > ok + rank = {"fail": 0, "warn": 1, "unknown": 2, "ok": 3} + + def _key(item): + return rank.get(item[1]["bucket"], 99) + + classified.sort(key=_key) + label, result = classified[0] + bucket = result["bucket"] + + if bucket == "unknown": + severity = policy["severity_mapping"].get(policy["unknown_action"], "info") + elif bucket == "ok": + severity = policy["severity_mapping"].get("ok", "info") + else: + severity = policy["severity_mapping"].get(bucket, "info") + + return { + "policy_applied": bucket, + "label": label, + "licenses": normalized_labels, + "reason": result["reason"], + "pattern_matched": result["pattern_matched"], + "severity": severity, + } diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/applications/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/applications/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/applications/test_runner_license_scan.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/applications/test_runner_license_scan.py new file mode 100644 index 000000000..618b602bd --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/test/applications/test_runner_license_scan.py @@ -0,0 +1,129 @@ +from unittest.mock import MagicMock, patch + +import pytest +import runpy + +from devsecops_engine_tools.engine_sca.engine_license.src.applications.runner_license_scan import ( + runner_engine_license, +) + + +def _make_devops_gateway(pipeline_name="svc"): + gw = MagicMock() + gw.get_variable.side_effect = lambda name: { + "pipeline_name": pipeline_name, + }.get(name, "") + return gw + + +def test_runner_engine_license_returns_findings_input_core_and_components(): + with patch( + "devsecops_engine_tools.engine_sca.engine_license.src.applications.runner_license_scan.init_engine_license" + ) as mock_init, patch( + "devsecops_engine_tools.engine_sca.engine_license.src.applications.runner_license_scan._build_findings_from_license_json" + ) as mock_findings: + mock_init.return_value = ("/abs/svc_LICENSE.json", ["c1"], {"THRESHOLD": {"COMPLIANCE": {"Critical": 2}}}) + mock_findings.return_value = [] + + config_tool = {"ENGINE_LICENSE": {"ENABLED": True}} + devops_gw = _make_devops_gateway() + + findings, input_core, sbom_components = runner_engine_license( + {"remote_config_repo": "r", "remote_config_branch": ""}, + config_tool, + None, + devops_gw, + None, + None, + ) + + mock_init.assert_called_once() + mock_findings.assert_called_once_with("/abs/svc_LICENSE.json") + + assert findings == [] + assert sbom_components == ["c1"] + + assert input_core.path_file_results == "/abs/svc_LICENSE.json" + assert input_core.totalized_exclusions == [] + assert input_core.scope_pipeline == "svc" + assert input_core.scope_service == "svc" + assert input_core.stage_pipeline == "Build" + assert input_core.custom_message_break_build == "License scan completed" + + assert input_core.threshold_defined is not None + assert input_core.threshold_defined.vulnerability.critical is None + assert input_core.threshold_defined.compliance.critical == 2 + + +def test_runner_engine_license_propagates_none_path(): + with patch( + "devsecops_engine_tools.engine_sca.engine_license.src.applications.runner_license_scan.init_engine_license" + ) as mock_init, patch( + "devsecops_engine_tools.engine_sca.engine_license.src.applications.runner_license_scan._build_findings_from_license_json" + ) as mock_findings: + mock_init.return_value = (None, None, {}) + mock_findings.return_value = [] + config_tool = {"ENGINE_LICENSE": {"ENABLED": True}} + devops_gw = _make_devops_gateway() + + findings, input_core, sbom_components = runner_engine_license( + {"remote_config_repo": "r", "remote_config_branch": ""}, + config_tool, + None, + devops_gw, + None, + None, + ) + + assert findings == [] + assert sbom_components is None + assert input_core.path_file_results is None + assert input_core.scope_pipeline == "svc" + assert input_core.threshold_defined.compliance.critical == 1 + + +def test_runner_license_main_block(): + with pytest.raises((TypeError, Exception)): + runpy.run_module( + "devsecops_engine_tools.engine_sca.engine_license.src.applications.runner_license_scan", + run_name="__main__", + alter_sys=True, + ) + + +def test_build_findings_from_license_json(tmp_path): + import json + from devsecops_engine_tools.engine_sca.engine_license.src.applications.runner_license_scan import ( + _build_findings_from_license_json, + ) + from devsecops_engine_tools.engine_core.src.domain.model.finding import Category + + license_data = { + "metadata": {"pipeline_name": "svc"}, + "dependencies": [ + {"name": "lodash", "version": "4.17.21", "licenses": ["MIT"], "policy_applied": "ok", "policy_reason": "compliant", "policy_pattern_matched": None, "license_matched": "MIT", "severity": "info"}, + {"name": "ngrx", "version": "1.0.0", "licenses": ["AGPL-3.0"], "policy_applied": "fail", "policy_reason": "matches FAIL pattern 'AGPL-*'", "policy_pattern_matched": "AGPL-*", "license_matched": "AGPL-3.0", "severity": "critical"}, + {"name": "biz-lib", "version": "2.0.0", "licenses": ["Apache-2.0", "EPL-2.0"], "policy_applied": "warn", "policy_reason": "matches WARN pattern 'EPL-*'", "policy_pattern_matched": "EPL-*", "license_matched": "EPL-2.0", "severity": "medium"}, + {"name": "no-lic", "version": "0.1.0", "licenses": [], "policy_applied": "unlicensed", "policy_reason": "no license detected", "policy_pattern_matched": None, "license_matched": "UNLICENSED", "severity": "info"}, + ], + } + path = tmp_path / "svc_LICENSE.json" + path.write_text(json.dumps(license_data)) + + findings = _build_findings_from_license_json(str(path)) + + assert len(findings) == 2 + assert findings[0].id == "AGPL-3.0-ngrx" + assert findings[0].severity == "critical" + assert findings[0].where == "ngrx:1.0.0" + assert findings[0].category == Category.COMPLIANCE + assert findings[0].module == "engine_license" + assert findings[1].id == "EPL-2.0-biz-lib" + assert findings[1].severity == "medium" + + +def test_build_findings_from_license_json_none_path(): + from devsecops_engine_tools.engine_sca.engine_license.src.applications.runner_license_scan import ( + _build_findings_from_license_json, + ) + assert _build_findings_from_license_json(None) == [] diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/model/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/model/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/model/test_context_license.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/model/test_context_license.py new file mode 100644 index 000000000..ee87441ff --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/model/test_context_license.py @@ -0,0 +1,36 @@ +from dataclasses import asdict + +from devsecops_engine_tools.engine_sca.engine_license.src.domain.model.context_license import ( + ContextLicense, +) + + +def test_context_license_instantiation_and_asdict(): + ctx = ContextLicense( + name="lodash", + version="4.17.21", + licenses=["MIT"], + policy_applied="ok", + policy_reason="Allowed by default", + policy_pattern_matched="", + severity="low", + ) + d = asdict(ctx) + assert d["name"] == "lodash" + assert d["licenses"] == ["MIT"] + assert d["severity"] == "low" + assert d["priority"] is None + + +def test_context_license_with_priority(): + ctx = ContextLicense( + name="ngrx", + version="1.0.0", + licenses=["AGPL-3.0"], + policy_applied="fail", + policy_reason="Matched AGPL-*", + policy_pattern_matched="AGPL-*", + severity="critical", + priority="very critical", + ) + assert ctx.priority == "very critical" diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/usecases/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/usecases/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/usecases/test_build_license_report.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/usecases/test_build_license_report.py new file mode 100644 index 000000000..0dab7f5d5 --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/test/domain/usecases/test_build_license_report.py @@ -0,0 +1,175 @@ +import json +import os +from unittest.mock import patch + +import pytest + +from devsecops_engine_tools.engine_sca.engine_license.src.domain.usecases.build_license_report import ( + BuildLicenseReport, +) + + +def _remote_config(): + return { + "LICENSE": { + "LICENSE_POLICY": { + "fail": ["AGPL-*"], + "warn": ["BUSL-*"], + "synonyms": {}, + "unlicensed_action": "warn", + "unknown_action": "ignore", + } + } + } + + +def _sbom_payload(components): + return { + "bomFormat": "CycloneDX", + "specVersion": "1.6", + "components": components, + } + + +def _write_sbom(tmp_path, payload): + path = tmp_path / "sbom.json" + path.write_text(json.dumps(payload)) + return str(path) + + +def test_process_writes_license_json_with_mixed_classifications(tmp_path): + components = [ + {"name": "lodash", "version": "4.17.21", "licenses": [{"license": {"id": "MIT"}}]}, + {"name": "ngrx", "version": "1.0.0", "licenses": [{"license": {"id": "AGPL-3.0"}}]}, + {"name": "biz-lib", "version": "2.0.0", "licenses": [{"license": {"id": "BUSL-1.1"}}]}, + {"name": "no-lic-pkg", "version": "0.1.0", "licenses": []}, + {"name": "weird-pkg", "version": "0.0.1", "licenses": [{"license": {"name": "Custom Proprietary License"}}]}, + ] + sbom_path = _write_sbom(tmp_path, _sbom_payload(components)) + + use_case = BuildLicenseReport(output_dir=str(tmp_path)) + out_path = use_case.process(sbom_path, _remote_config(), "svc") + + assert out_path is not None + assert os.path.exists(out_path) + assert out_path.endswith("svc_LICENSE.json") + + with open(out_path) as fh: + report = json.load(fh) + + md = report["metadata"] + assert md["pipeline_name"] == "svc" + assert md["tool"] == "CDXGEN" + assert md["policy_used"] == _remote_config()["LICENSE"]["LICENSE_POLICY"] + assert md["summary"]["total_dependencies"] == 5 + assert md["summary"]["ok"] == 1 + assert md["summary"]["fail"] == 1 + assert md["summary"]["warn"] == 1 + assert md["summary"]["unlicensed"] == 1 + assert md["summary"]["unknown"] == 1 + + deps = report["dependencies"] + assert len(deps) == 5 + names = [d["name"] for d in deps] + assert names == ["lodash", "ngrx", "biz-lib", "no-lic-pkg", "weird-pkg"] + + ngrx = next(d for d in deps if d["name"] == "ngrx") + assert ngrx["policy_applied"] == "fail" + assert ngrx["policy_pattern_matched"] == "AGPL-*" + assert "AGPL-*" in ngrx["policy_reason"] + assert ngrx["severity"] == "critical" + + biz_lib = next(d for d in deps if d["name"] == "biz-lib") + assert biz_lib["severity"] == "medium" + + no_lic = next(d for d in deps if d["name"] == "no-lic-pkg") + assert no_lic["policy_applied"] == "unlicensed" + assert no_lic["licenses"] == [] + + +def test_process_dual_license_highest_risk_wins(tmp_path): + components = [ + {"name": "dual-pkg", "version": "1.0", "licenses": [{"license": {"id": "AGPL-3.0"}}, {"license": {"id": "MIT"}}]}, + ] + sbom_path = _write_sbom(tmp_path, _sbom_payload(components)) + + out = BuildLicenseReport(output_dir=str(tmp_path)).process( + sbom_path, _remote_config(), "svc" + ) + + with open(out) as fh: + report = json.load(fh) + dep = report["dependencies"][0] + assert dep["policy_applied"] == "fail" + assert sorted(dep["licenses"]) == ["AGPL-3.0", "MIT"] + + +def test_process_metadata_policy_used_is_deep_copy(tmp_path): + config = _remote_config() + sbom_path = _write_sbom( + tmp_path, + _sbom_payload([{"name": "x", "version": "1", "licenses": [{"license": {"id": "MIT"}}]}]), + ) + out = BuildLicenseReport(output_dir=str(tmp_path)).process( + sbom_path, config, "svc" + ) + + with open(out) as fh: + report = json.load(fh) + report["metadata"]["policy_used"]["fail"].append("MUTATED") + assert config["LICENSE"]["LICENSE_POLICY"]["fail"] == ["AGPL-*"] + + +def test_process_returns_none_when_remote_config_missing(tmp_path): + sbom_path = _write_sbom(tmp_path, _sbom_payload([])) + assert ( + BuildLicenseReport(output_dir=str(tmp_path)).process( + sbom_path, {}, "svc" + ) + is None + ) + + +def test_process_returns_none_when_sbom_missing(tmp_path): + out = BuildLicenseReport(output_dir=str(tmp_path)).process( + str(tmp_path / "does_not_exist.json"), _remote_config(), "svc" + ) + assert out is None + + +def test_process_returns_none_when_sbom_path_is_empty(tmp_path): + out = BuildLicenseReport(output_dir=str(tmp_path)).process( + "", _remote_config(), "svc" + ) + assert out is None + + +def test_process_returns_none_when_sbom_invalid_json(tmp_path): + bad = tmp_path / "bad.json" + bad.write_text("not-json{{{" ) + out = BuildLicenseReport(output_dir=str(tmp_path)).process( + str(bad), _remote_config(), "svc" + ) + assert out is None + + +def test_process_handles_empty_components(tmp_path): + sbom_path = _write_sbom(tmp_path, {"components": []}) + out = BuildLicenseReport(output_dir=str(tmp_path)).process( + sbom_path, _remote_config(), "svc" + ) + with open(out) as fh: + report = json.load(fh) + assert report["dependencies"] == [] + assert report["metadata"]["summary"]["total_dependencies"] == 0 + + +def test_default_output_dir_is_cwd(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + sbom_path = _write_sbom( + tmp_path, + _sbom_payload([{"name": "x", "version": "1", "licenses": [{"license": {"id": "MIT"}}]}]), + ) + out = BuildLicenseReport().process(sbom_path, _remote_config(), "svc") + assert out is not None + assert os.path.dirname(out) == str(tmp_path) diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/driven_adapters/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/driven_adapters/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/driven_adapters/license_scan/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/driven_adapters/license_scan/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/driven_adapters/license_scan/test_license_scan_manager.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/driven_adapters/license_scan/test_license_scan_manager.py new file mode 100644 index 000000000..1ab769b5e --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/driven_adapters/license_scan/test_license_scan_manager.py @@ -0,0 +1,30 @@ +import json + +from devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.driven_adapters.license_scan.license_scan_manager import ( + LicenseScanManager, +) + + +def test_get_license_context_from_results_success(tmp_path): + license_data = { + "metadata": {"pipeline_name": "svc"}, + "dependencies": [ + {"name": "lodash", "version": "4.17.21", "licenses": ["MIT"], "policy_applied": "ok", "policy_reason": "compliant", "policy_pattern_matched": None, "severity": "info"}, + {"name": "ngrx", "version": "1.0.0", "licenses": ["AGPL-3.0"], "policy_applied": "fail", "policy_reason": "matches FAIL pattern", "policy_pattern_matched": "AGPL-*", "severity": "critical"}, + {"name": "jakarta.servlet-api", "version": "6.1.0", "licenses": ["EPL-2.0", "GPL-2.0"], "policy_applied": "warn", "policy_reason": "matches WARN pattern", "policy_pattern_matched": "EPL-*", "severity": "medium"}, + {"name": "no-lic", "version": "0.1.0", "licenses": [], "policy_applied": "unlicensed", "policy_reason": "no license", "policy_pattern_matched": None, "severity": "info"}, + ], + } + path = tmp_path / "svc_LICENSE.json" + path.write_text(json.dumps(license_data)) + + manager = LicenseScanManager() + result = manager.get_license_context_from_results(str(path)) + + # Only fail and warn appear + assert len(result) == 2 + assert result[0].name == "ngrx" + assert result[0].severity == "critical" + assert result[1].name == "jakarta.servlet-api" + assert result[1].severity == "medium" + assert result[0].priority is None diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/entry_points/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/entry_points/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/entry_points/test_entry_point_tool.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/entry_points/test_entry_point_tool.py new file mode 100644 index 000000000..0b6524772 --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/entry_points/test_entry_point_tool.py @@ -0,0 +1,166 @@ +from unittest.mock import MagicMock, patch + +from devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.entry_points.entry_point_tool import ( + init_engine_license, +) + + +def _make_devops_gateway(pipeline_name="svc"): + gw = MagicMock() + gw.get_variable.side_effect = lambda name: { + "pipeline_name": pipeline_name, + "branch_tag": "main", + }.get(name, "") + return gw + + +def _make_remote_gw(remote_cfg): + gw = MagicMock() + gw.get_remote_config.return_value = remote_cfg + return gw + + +def _config_tool(): + return { + "ENGINE_LICENSE": {"ENABLED": True}, + "SBOM_MANAGER": {"ENABLED": True}, + } + + +def _remote_config(): + return { + "LICENSE": { + "LICENSE_POLICY": { + "fail": [], + "warn": [], + "synonyms": {}, + "unlicensed_action": "ignore", + "unknown_action": "ignore", + } + } + } + + +@patch( + "devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.entry_points.entry_point_tool.BuildLicenseReport" +) +@patch( + "devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.entry_points.entry_point_tool.os.path.exists" +) +def test_init_engine_license_happy_path(mock_exists, mock_builder): + mock_exists.return_value = True + mock_builder.return_value.process.return_value = "/abs/svc_LICENSE.json" + + devops_gw = _make_devops_gateway() + remote_gw = _make_remote_gw(_remote_config()) + tool_sbom = MagicMock() + tool_sbom.get_components.return_value = ["c1", "c2"] + + license_path, sbom_components, remote_config = init_engine_license( + devops_gw, + remote_gw, + {"remote_config_repo": "r", "remote_config_branch": ""}, + _config_tool(), + tool_sbom, + ) + + assert license_path == "/abs/svc_LICENSE.json" + assert sbom_components == ["c1", "c2"] + assert remote_config == _remote_config() + tool_sbom.get_components.assert_called_once() + mock_builder.return_value.process.assert_called_once_with( + "svc_SBOM.json", _remote_config(), "svc" + ) + + +@patch( + "devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.entry_points.entry_point_tool.os.path.exists", + return_value=False, +) +def test_init_engine_license_returns_none_when_to_scan_missing(mock_exists): + devops_gw = _make_devops_gateway() + remote_gw = _make_remote_gw(_remote_config()) + tool_sbom = MagicMock() + + license_path, sbom_components, remote_config = init_engine_license( + devops_gw, + remote_gw, + {"remote_config_repo": "r", "remote_config_branch": "", "folder_path": "/missing"}, + _config_tool(), + tool_sbom, + ) + + assert license_path is None + assert sbom_components is None + assert remote_config == _remote_config() + tool_sbom.get_components.assert_not_called() + + +@patch( + "devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.entry_points.entry_point_tool.os.path.exists" +) +def test_init_engine_license_returns_none_when_sbom_missing_after_generation(mock_exists): + mock_exists.side_effect = [True, False] + devops_gw = _make_devops_gateway() + remote_gw = _make_remote_gw(_remote_config()) + tool_sbom = MagicMock() + tool_sbom.get_components.return_value = ["c"] + + license_path, sbom_components, remote_config = init_engine_license( + devops_gw, + remote_gw, + {"remote_config_repo": "r", "remote_config_branch": ""}, + _config_tool(), + tool_sbom, + ) + + assert license_path is None + assert sbom_components == ["c"] + assert remote_config == _remote_config() + + +@patch( + "devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.entry_points.entry_point_tool.os.path.exists", + return_value=True, +) +def test_init_engine_license_returns_none_when_tool_sbom_missing(mock_exists): + devops_gw = _make_devops_gateway() + remote_gw = _make_remote_gw(_remote_config()) + + license_path, sbom_components, remote_config = init_engine_license( + devops_gw, + remote_gw, + {"remote_config_repo": "r", "remote_config_branch": ""}, + _config_tool(), + None, + ) + + assert license_path is None + assert sbom_components is None + assert remote_config == _remote_config() + + +@patch( + "devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.entry_points.entry_point_tool.BuildLicenseReport" +) +@patch( + "devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.entry_points.entry_point_tool.os.path.exists", + return_value=True, +) +def test_init_engine_license_returns_none_when_build_report_fails(mock_exists, mock_builder): + mock_builder.return_value.process.return_value = None + devops_gw = _make_devops_gateway() + remote_gw = _make_remote_gw(_remote_config()) + tool_sbom = MagicMock() + tool_sbom.get_components.return_value = ["c"] + + license_path, sbom_components, remote_config = init_engine_license( + devops_gw, + remote_gw, + {"remote_config_repo": "r", "remote_config_branch": ""}, + _config_tool(), + tool_sbom, + ) + assert license_path is None + assert sbom_components == ["c"] + assert remote_config == _remote_config() diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/helpers/__init__.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/helpers/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/helpers/test_license_policy.py b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/helpers/test_license_policy.py new file mode 100644 index 000000000..c7711e393 --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sca/engine_license/test/infrastructure/helpers/test_license_policy.py @@ -0,0 +1,191 @@ +from devsecops_engine_tools.engine_sca.engine_license.src.infrastructure.helpers.license_policy import ( + build_policy_from_remote_config, + classify_package, + get_value, + looks_like_spdx_id, + validate_action, + _DEFAULT_SEVERITY_BY_ACTION, +) + + +def _policy(**overrides): + """Helper: build a default-flavoured policy and apply overrides.""" + base = { + "fail": ["AGPL-*", "SSPL-*"], + "warn": ["BUSL-*", "EPL-*", "LGPL-3.0*"], + "synonyms": {}, + "unlicensed_action": "ignore", + "unknown_action": "ignore", + "severity_mapping": dict(_DEFAULT_SEVERITY_BY_ACTION), + } + base.update(overrides) + return base + + +def test_get_value_returns_first_non_none(): + obj = {"a": None, "B": "value", "c": "other"} + assert get_value(obj, "a", "B", "c") == "value" + assert get_value({}, "x", default="d") == "d" + assert get_value(None, "x", default="d") == "d" + + +def test_validate_action_clamps_to_known_set(): + assert validate_action("FAIL", "ignore") == "fail" + assert validate_action(" Warn ", "ignore") == "warn" + assert validate_action("bogus", "ignore") == "ignore" + + +def test_looks_like_spdx_id(): + assert looks_like_spdx_id("MIT") + assert looks_like_spdx_id("Apache-2.0") + assert looks_like_spdx_id("GPL-3.0+") + assert not looks_like_spdx_id("") + assert not looks_like_spdx_id("MIT License") + assert not looks_like_spdx_id("a" * 80) + +def test_build_policy_returns_none_when_remote_config_missing(): + assert build_policy_from_remote_config(None) is None + assert build_policy_from_remote_config({}) is None + assert build_policy_from_remote_config({"LICENSE": {}}) is None + + +def test_build_policy_normalises_lists_and_actions(): + raw = { + "LICENSE": { + "LICENSE_POLICY": { + "fail": ["AGPL-*"], + "warn": ["BUSL-*"], + "synonyms": {"BSD": "BSD-3-Clause"}, + "unlicensed_action": "WARN", + "unknown_action": "Bogus", + } + } + } + policy = build_policy_from_remote_config(raw) + assert policy["fail"] == ["AGPL-*"] + assert policy["warn"] == ["BUSL-*"] + assert policy["synonyms"] == {"BSD": "BSD-3-Clause"} + assert policy["unlicensed_action"] == "warn" + assert policy["unknown_action"] == "ignore" + + +def test_build_policy_handles_non_list_fail_warn(): + raw = {"LICENSE": {"LICENSE_POLICY": {"fail": "not-a-list", "warn": None}}} + policy = build_policy_from_remote_config(raw) + assert policy["fail"] == [] + assert policy["warn"] == [] + + +def test_build_policy_default_severity_mapping(): + raw = {"LICENSE": {"LICENSE_POLICY": {"fail": [], "warn": []}}} + policy = build_policy_from_remote_config(raw) + assert policy["severity_mapping"] == _DEFAULT_SEVERITY_BY_ACTION + + +def test_build_policy_severity_mapping_override(): + raw = { + "LICENSE": { + "LICENSE_POLICY": { + "fail": ["AGPL-*"], + "warn": [], + "severity_mapping": {"fail": "high"}, + } + } + } + policy = build_policy_from_remote_config(raw) + assert policy["severity_mapping"]["fail"] == "high" + assert policy["severity_mapping"]["warn"] == _DEFAULT_SEVERITY_BY_ACTION["warn"] + + result = classify_package([{"id": "AGPL-3.0"}], policy) + assert result["policy_applied"] == "fail" + assert result["severity"] == "high" + +def test_classify_package_compliant_spdx(): + result = classify_package([{"id": "MIT"}], _policy()) + assert result["policy_applied"] == "ok" + assert result["label"] == "MIT" + assert result["licenses"] == ["MIT"] + assert result["pattern_matched"] is None + assert result["severity"] == "info" + + +def test_classify_package_fail_pattern(): + result = classify_package([{"id": "AGPL-3.0"}], _policy()) + assert result["policy_applied"] == "fail" + assert result["pattern_matched"] == "AGPL-*" + assert "AGPL-*" in result["reason"] + assert result["severity"] == "critical" + + +def test_classify_package_warn_pattern(): + result = classify_package([{"id": "BUSL-1.1"}], _policy()) + assert result["policy_applied"] == "warn" + assert result["pattern_matched"] == "BUSL-*" + assert result["severity"] == "medium" + + +def test_classify_package_dual_license_highest_risk_wins(): + result = classify_package( + [{"id": "AGPL-3.0"}, {"id": "MIT"}], _policy() + ) + assert result["policy_applied"] == "fail" + assert sorted(result["licenses"]) == ["AGPL-3.0", "MIT"] + + +def test_classify_package_dual_warn_and_fail_picks_fail(): + result = classify_package( + [{"id": "AGPL-3.0"}, {"id": "BUSL-1.1"}], _policy() + ) + assert result["policy_applied"] == "fail" + + +def test_classify_package_synonyms_resolved(): + policy = _policy(synonyms={"BSD": "BSD-3-Clause"}) + result = classify_package([{"id": "BSD"}], policy) + assert result["policy_applied"] == "ok" + assert result["label"] == "BSD-3-Clause" + + +def test_classify_package_unlicensed_ignore_severity_info(): + result = classify_package([], _policy(unlicensed_action="ignore")) + assert result["policy_applied"] == "unlicensed" + assert result["label"] == "UNLICENSED" + assert result["severity"] == "info" + + +def test_classify_package_unlicensed_warn_severity_medium(): + result = classify_package([], _policy(unlicensed_action="warn")) + assert result["policy_applied"] == "unlicensed" + assert result["label"] == "UNLICENSED" + assert result["severity"] == "medium" + + +def test_classify_package_unknown_label_ignore_severity_info(): + result = classify_package( + [{"name": "Custom Proprietary License"}], + _policy(unknown_action="ignore"), + ) + assert result["policy_applied"] == "unknown" + assert result["severity"] == "info" + + +def test_classify_package_unknown_label_fail_severity_critical(): + result = classify_package( + [{"name": "Custom Proprietary License"}], + _policy(unknown_action="fail"), + ) + assert result["policy_applied"] == "unknown" + assert result["severity"] == "critical" + + +def test_classify_package_with_none_policy(): + result = classify_package([{"id": "MIT"}], None) + assert result["policy_applied"] == "unknown" + assert result["label"] == "UNKNOWN" + assert result["severity"] == "info" + + +def test_severity_by_action_defaults(): + assert _DEFAULT_SEVERITY_BY_ACTION["fail"] == "critical" + assert _DEFAULT_SEVERITY_BY_ACTION["warn"] == "medium" + assert _DEFAULT_SEVERITY_BY_ACTION["ok"] == "info"