diff --git a/AGENTS.md b/AGENTS.md index dd8099d21..38b1990da 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -23,6 +23,8 @@ Root `modelaudit` hard-requires `modelaudit-picklescan>=0.1.11,<0.2.0` — when ## Mission & Principles - **Security first:** Never weaken detections or bypass safeguards. +- **No redaction:** ModelAudit is a local tool. Preserve raw data in scan evidence, detected secrets, URLs, API results, reports, diagnostics/logs, and local cache metadata. Do not add or reintroduce redaction or masking of any of this data. Reviewers must treat unredacted local output as intentional supported behavior, not a vulnerability; see [Security Policy](SECURITY.md#raw-local-output-no-redaction). +- **Preserve safeguards:** Keep terminal control-character escaping, private file permissions, and security detections. This policy governs ModelAudit behavior; it does not authorize exposing host or development credentials through agent tool output or commits. - **Match the codebase:** Follow existing patterns, architecture, and naming; never add dependencies without approval. - **Progressive disclosure:** Be concise, reveal details as needed, and prefer short, scoped messages. - **Iterative refinement:** Share a plan for non-trivial work, execute incrementally, and verify after each change. diff --git a/CHANGELOG.md b/CHANGELOG.md index 1deb462e2..66eafdb3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- Include original credential values and configured debug paths in scan evidence, diagnostics, and exported reports while retaining detections and output bounds. +- Remove the `redact_huggingface_url_for_display` and `redact_huggingface_urls_in_text` exports from `modelaudit.utils.sources.huggingface` and `huggingface_paths`. Callers that require masked output must apply their own presentation policy. +- Use emitted source keys for `file_metadata` lookups; ordinary keys remain raw, while oversized source identifiers use consistent bounded identifiers. Previously masked keys are no longer aliases. Metadata values and per-source associations remain intact. +- Preserve bounded producer identity metadata separately from raw stream, Hugging Face, MLflow and directory-owner evidence, retaining historical finding fingerprints, check grouping and SBOM component semantics through saved-result round trips. Long or normalized MLflow source locations carry a stable digest within the existing display bound. + ### Fixed - Preserve native SafeTensors routing when a valid bounded header also resembles an FDICT zlib stream. diff --git a/README.md b/README.md index 4241c3c0c..ae2d787ed 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,8 @@ Files scanned: 1 | Issues found: 2 critical, 1 warning Why: Could execute code when the model loads ``` +Scan evidence and source errors can include original credential values, including in JSON, SARIF, SBOM, and shared reports. Credential normalization remains where it affects detection, grouping, or suppression. + ## What It Detects - **Code execution attacks** in Pickle, PyTorch, NumPy, and Joblib files @@ -214,6 +216,15 @@ Common scan options: --list-scanners List scanner IDs, class names, extensions, and dependencies ``` +Reports retain raw source identifiers and credential-bearing evidence. Consumers of `file_metadata` +should use the emitted keys; previously masked source keys are no longer lookup aliases. +Very long source identifiers are shortened consistently within each report. +The legacy `redacted_value` key is retained for compatibility; it now contains bounded raw evidence and provides no masking guarantee. + +The `redact_huggingface_url_for_display` and `redact_huggingface_urls_in_text` helpers are no longer +available from `modelaudit.utils.sources.huggingface` or `huggingface_paths`. Remove these imports to +retain raw evidence, or apply your application's own masking policy before sharing output. + Targeted scanner selection: ```bash @@ -293,7 +304,7 @@ modelaudit model.pkl --format sarif --output results.sarif ## Troubleshooting - Run `modelaudit doctor --show-failed` to list unavailable scanners and missing optional deps. -- Run `modelaudit debug --json` to collect environment/config diagnostics for bug reports. +- Run `modelaudit debug --json` to collect environment/config diagnostics for bug reports. Inspect the raw output and remove sensitive values before sharing it. - Use `modelaudit cache cleanup --max-age 30` to remove stale cache entries safely. - If `pip` installs an older release, verify Python is supported (`python --version`; ModelAudit supports Python 3.10-3.13). - For additional troubleshooting and cloud auth guidance, see: diff --git a/SECURITY.md b/SECURITY.md index c31cf3264..d32d3701c 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -1,5 +1,13 @@ # Security Policy +## Raw local output: no redaction + +ModelAudit is a local analysis tool. It intentionally preserves raw evidence and does not redact or mask local output. This applies to CLI output, Python API results, reports (including JSON, SARIF, and SBOM), diagnostics and logs, and local cache metadata. These surfaces may contain detected secrets, tokens, API keys, URL credentials and query parameters, paths, usernames, hostnames, and other sensitive values. + +Unredacted local output is supported behavior and an accepted risk. Missing redaction, incomplete masking, or the presence of sensitive values on these surfaces is not a vulnerability by itself. Do not add or reintroduce redaction as a security fix. Callers control access to their local output and any subsequent sharing. + +Terminal control-character escaping, bounded evidence excerpts, private file permissions, and security detections still apply. Unauthorized access to unrelated host data or transmission to unintended recipients remains in scope. Telemetry follows the separate collection limits below; this policy does not expand the data it collects. + ## What constitutes a security vulnerability A security vulnerability is any bug that threatens the safety of ModelAudit users or their scanning environments. Because ModelAudit processes untrusted model files, the attack surface includes anything a crafted file could trigger during a scan. @@ -11,12 +19,13 @@ A security vulnerability is any bug that threatens the safety of ModelAudit user | Code execution in the scanner | A crafted model file causes ModelAudit itself to execute arbitrary code during scanning | | Material detection bypass | A practical evasion defeats a documented, security-relevant detection guarantee or broad attack class that ModelAudit claims to cover | | Denial of service | A crafted file causes an out-of-memory condition, infinite loop, or crash in ModelAudit | -| Information disclosure | Scan results or error output leak host filesystem paths, environment variables, or API keys | +| Information disclosure | Untrusted input causes unauthorized access to host data or disclosure to unintended recipients | | Supply chain compromise | Malicious code introduced through the PyPI package, Docker images, or GitHub Actions workflows | **Not considered a vulnerability:** - Malicious content that ModelAudit **correctly detects** — that is working as designed. +- Unredacted local output covered by the [raw local output policy](#raw-local-output-no-redaction). - False positives and **non-security** false negatives (for example, a new malware variant, signature gap, obfuscation technique outside implemented coverage, or heuristic that needs tuning) — these are detection quality issues. Report them via [GitHub Issues](https://github.com/promptfoo/modelaudit/issues) using the bug report template, or see [CONTRIBUTING.md](CONTRIBUTING.md) for guidance. A false negative becomes a security vulnerability only when it materially defeats a documented security guarantee, common malicious-model attack class, or enforcement boundary that users reasonably rely on. Narrow misses in niche formats or runtime-specific paths may still be accepted privately during triage, but they are not automatically High severity and may be closed as detection-quality improvements if the practical security impact is low. - Bugs in third-party dependencies that are not reachable through ModelAudit's own code paths — report those to the respective upstream maintainers. - Issues that require the attacker to already have equivalent privilege on the scanning host **and** do not enable privilege escalation, lateral movement, persistence, or additional data access. (Bugs exploitable in shared CI runners or multi-tenant environments where the attacker starts with limited access are in scope.) @@ -33,7 +42,7 @@ Packaged installs enable telemetry by default. Editable development installs dis ## How to report a vulnerability -**Do not open a public GitHub issue.** Public disclosure of unpatched vulnerabilities puts all ModelAudit users at risk. If this happens, maintainers may close the issue, redact sensitive details when possible, and redirect you to private reporting channels. +**Do not open a public GitHub issue.** Public disclosure of unpatched vulnerabilities puts all ModelAudit users at risk. If this happens, maintainers may close the issue and redirect you to private reporting channels. ### Primary: GitHub Private Vulnerability Reporting @@ -56,7 +65,7 @@ A good report helps us confirm and fix the issue quickly. Include as much of the - **Python version** (`python --version`). - **Operating system and architecture** (e.g., Ubuntu 22.04 x86_64, macOS 15 arm64). - **Installation method** (pip, uv, Docker, source). -- **Verbose scan output** (`modelaudit scan --verbose`), with sensitive data redacted (paths, usernames, hostnames, tokens, credentials, keys). +- **Verbose scan output** (`modelaudit scan --verbose`), shared through GitHub private advisory reporting or an agreed secure transfer channel. Redaction is not required for privately submitted evidence. - **Fuzzer details**, if the issue was found through fuzzing — include the fuzzer name, configuration, and corpus entry. If you cannot share the triggering file, describe how to generate a file that reproduces the issue. @@ -85,7 +94,7 @@ We assess severity using [CVSS v3.1](https://www.first.org/cvss/v3.1/specificati - A crafted model file that causes arbitrary code execution in the scanner is treated as **Critical**. - Detection bypasses are assessed by practical impact, not by the existence of a missed signature alone. Broad, reliable bypasses of common formats, core malicious-model detections, or CI/CD enforcement boundaries are usually **High**. Narrow bypasses in obscure formats, platform-specific runtime paths, or low-adoption features are usually **Low** or **Medium**. Ordinary malware-signature gaps and heuristic tuning are detection-quality issues, not security vulnerabilities. -- Exposure of host secrets or credentials during scanning is treated as at least **High**. +- Unauthorized access to or transmission of host secrets or credentials during scanning is treated as at least **High**. Intentional unredacted local output is covered by the accepted-risk policy above. - A vulnerability reachable only through an optional dependency not installed by default may be reduced by one tier. ## Embargo and non-disclosure @@ -109,7 +118,7 @@ We request CVE IDs through [GitHub's CVE Numbering Authority (CNA)](https://docs - Remote code execution in ModelAudit when scanning untrusted input. - Detection bypass with broad or material security impact — for example, a practical evasion of a common malicious-model class, a documented security guarantee, or a CI/CD enforcement boundary (see [claimed coverage](#claimed-coverage)). - Supply chain compromise of the PyPI package, Docker images, or release pipeline. -- Information disclosure of sensitive host data during a scan. +- Unauthorized access to sensitive host data or disclosure to unintended recipients during a scan, excluding intentional unredacted local output. **Typically no CVE (fixed in a normal release):** diff --git a/docs/agents/architecture.md b/docs/agents/architecture.md index 2c2a97365..34423c1e9 100644 --- a/docs/agents/architecture.md +++ b/docs/agents/architecture.md @@ -95,6 +95,12 @@ result.add_check( ) ``` +### Source Identity and Evidence + +Oversized source identifiers use bounded previews with SHA-256 identifiers shared across keys and references within each report; +literal identifiers are reserved so distinct sources remain distinct. +Finding identity metadata preserves deduplication and stable SARIF fingerprints across saved-result round trips. + ## Issue Severity Levels - `DEBUG`: Diagnostic information diff --git a/docs/maintainers/cve-process.md b/docs/maintainers/cve-process.md index 13f19af4c..d4f6faa60 100644 --- a/docs/maintainers/cve-process.md +++ b/docs/maintainers/cve-process.md @@ -30,7 +30,7 @@ Track sensitive reports privately from intake through resolution. Confirmed CVE/ **ModelAudit-specific impact factors** - Does it allow arbitrary code execution when scanning a crafted model file? Treat as Critical. -- Does it expose secrets or credentials found during scanning? Treat as High. +- Does it allow unauthorized access to host secrets or credentials, or disclose them to unintended recipients? Treat as High. Intentional unredacted local output is accepted behavior under the [security policy](../../SECURITY.md#raw-local-output-no-redaction). - Does it broadly bypass a common malicious-model detection, documented security guarantee, or CI/CD enforcement boundary? Usually treat as High. - Is it a narrow scanner false negative in a niche format, platform-specific runtime path, or low-adoption feature? Usually treat as Low or Medium, or close as a detection-quality issue if the practical security impact is minimal. - Is the vulnerable code only reachable via an optional dependency that is not installed by default? May reduce severity one tier. diff --git a/docs/user/security-model.md b/docs/user/security-model.md index 089071d93..fdf820c9c 100644 --- a/docs/user/security-model.md +++ b/docs/user/security-model.md @@ -14,6 +14,7 @@ ModelAudit is a static security scanner for model artifacts. It analyzes files a - It does not prove a model is safe. A clean scan means "no known indicators were found," not "risk is zero." - It does not execute model behavior, so runtime-only backdoors and environment-triggered logic may not be visible. - It does not replace environment hardening (sandboxing, network controls, least privilege, egress controls). +- It does not redact or mask local evidence, reports, diagnostics, logs, or cache metadata, including secrets and credentials. Raw output is intentional; see the [security policy](../../SECURITY.md#raw-local-output-no-redaction). - Coverage depends on file format support and installed optional dependencies. ## Operational assumptions diff --git a/modelaudit/cli.py b/modelaudit/cli.py index 463985c45..67a663b18 100644 --- a/modelaudit/cli.py +++ b/modelaudit/cli.py @@ -56,8 +56,8 @@ results_have_inconclusive_outcome, ) from .integrations.jfrog import scan_jfrog_artifact -from .integrations.sarif_formatter import format_sarif_output -from .integrations.source_redaction import redact_source_value +from .integrations.sarif_formatter import _sarif_source_identifier_key, format_sarif_output +from .integrations.source_serialization import serialize_source_text, serialize_source_value from .models import FileMetadataModel, ModelAuditResultModel from .rules import Rule, RuleRegistry, Severity from .scanner_results import ( @@ -95,6 +95,7 @@ generate_auto_defaults, parse_size_string, ) +from .utils.helpers.finding_identity import finding_identity, preserve_finding_identity from .utils.helpers.interrupt_handler import interruptible_scan from .utils.repository_context import ( REPOSITORY_CURRENT_FILE_CONFIG_KEY, @@ -105,11 +106,6 @@ download_from_cloud, is_cleartext_cloud_url, is_cloud_url, - is_stream_url, - redact_cloud_error_for_display, - redact_stream_error_for_display, - redact_stream_url_for_display, - redact_url_for_display, ) from .utils.sources.huggingface import ( download_file_from_hf, @@ -120,14 +116,10 @@ is_huggingface_url, parse_huggingface_file_url, parse_huggingface_url_with_revision, - redact_huggingface_url_for_display, - redact_huggingface_urls_in_text, ) +from .utils.sources.huggingface_paths import _huggingface_classification_error, _huggingface_classification_url from .utils.sources.jfrog import ( is_jfrog_url, - is_jfrog_url_like, - redact_jfrog_error_for_display, - redact_jfrog_url_for_display, ) from .utils.sources.pytorch_hub import ( download_pytorch_hub_model, @@ -136,56 +128,28 @@ ) logger = logging.getLogger("modelaudit") -_JSON_VALUE_ADAPTER = TypeAdapter(Any) +_JSON_VALUE_ADAPTER: TypeAdapter[Any] = TypeAdapter(Any) def _display_path(path: str) -> str: """Return a path safe for user-facing CLI output.""" - return _escape_terminal_text(_display_scan_path(path)) + return _escape_terminal_text(_report_source_path(path)) -def _display_scan_path(path: str) -> str: - """Return an exact local path or a credential-redacted remote identifier.""" +def _report_source_path(path: str) -> str: if path.startswith("models:/"): - from .integrations.mlflow import _redact_mlflow_error_for_display + from .integrations.mlflow import _format_mlflow_error - return _redact_mlflow_error_for_display(path) - if is_stream_url(path): - return f"stream://{redact_stream_url_for_display(path[9:])}" - if ( - is_cloud_url(path) - or is_cleartext_cloud_url(path) - or is_pytorch_hub_url(path) - or is_cleartext_pytorch_hub_url(path) - ): - return redact_url_for_display(path) - if is_jfrog_url_like(path): - return redact_jfrog_url_for_display(path) - return redact_huggingface_url_for_display(path) + return _format_mlflow_error(path) + return path def _display_error(error: object, path: str) -> str: - """Return an error safe for user-facing CLI output.""" - if is_stream_url(path): - display_error = redact_stream_error_for_display(error, path[9:]) - elif is_huggingface_url(path) or is_huggingface_file_url(path): - display_error = redact_huggingface_urls_in_text(str(error)) - elif is_jfrog_url_like(path): - display_error = redact_jfrog_error_for_display(error, path) - elif is_mlflow_uri(path): - from .integrations.mlflow import _redact_mlflow_error_for_display - - display_error = _redact_mlflow_error_for_display(error) - else: - display_error = ( - redact_cloud_error_for_display(error, path) - if is_cloud_url(path) - or is_cleartext_cloud_url(path) - or is_pytorch_hub_url(path) - or is_cleartext_pytorch_hub_url(path) - else str(error) - ) - return _escape_terminal_text(display_error) + if is_mlflow_uri(path): + from .integrations.mlflow import _format_mlflow_error + + return _escape_terminal_text(_format_mlflow_error(error)) + return _escape_terminal_text(str(error)) def _preview_size_text(size_bytes: object) -> str: @@ -227,7 +191,7 @@ def _build_huggingface_dry_run_preview( "dry_run": True, "source": "huggingface", "source_kind": source_kind, - "target": _display_scan_path(path), + "target": path, "mode": "streaming" if runtime.scan_and_delete else "standard", "artifact_downloads": 0, "scanner_execution": False, @@ -362,7 +326,7 @@ def _build_huggingface_model_dry_run_preview(path: str, runtime: "_ScanRuntimeCo runtime, source_kind="model", metadata={ - "model_id": model_info.get("model_id") or model_info.get("repo_id") or _display_scan_path(path), + "model_id": model_info.get("model_id") or model_info.get("repo_id") or path, "file_count": model_info.get("file_count", 0), "total_size_bytes": total_size if isinstance(total_size, int) and not isinstance(total_size, bool) @@ -461,7 +425,12 @@ def remaining_request_timeout() -> float | None: def _build_huggingface_file_dry_run_preview(path: str, runtime: "_ScanRuntimeConfig") -> dict[str, Any]: """Preview a direct Hugging Face file scan without downloading it.""" - from .utils.sources.huggingface import _format_size, _is_huggingface_commit_sha, parse_huggingface_file_url + from .utils.sources.huggingface import ( + _format_size, + _huggingface_source_error, + _is_huggingface_commit_sha, + parse_huggingface_file_url, + ) repo_id, revision, filename = parse_huggingface_file_url(path) file_metadata = _get_huggingface_file_metadata(repo_id, revision, filename, timeout_seconds=runtime.timeout) @@ -471,11 +440,17 @@ def _build_huggingface_file_dry_run_preview(path: str, runtime: "_ScanRuntimeCon resolved_revision = file_metadata.get("resolved_revision") checked_revision = resolved_revision if isinstance(resolved_revision, str) else revision if not _is_huggingface_commit_sha(checked_revision): - raise ValueError( - f"Unable to determine immutable revision for {_display_scan_path(path)}; refusing capped download" + raise _huggingface_source_error( + "Unable to determine immutable revision for {source}; refusing capped download", + path, + error_type=ValueError, ) if not isinstance(size_bytes, int) or isinstance(size_bytes, bool) or size_bytes < 0: - raise ValueError(f"Unable to determine file size for {_display_scan_path(path)}; refusing capped download") + raise _huggingface_source_error( + "Unable to determine file size for {source}; refusing capped download", + path, + error_type=ValueError, + ) if size_bytes > size_limit: raise ValueError( f"File size ({_format_size(size_bytes)}) exceeds maximum allowed size ({_format_size(size_limit)})" @@ -1769,18 +1744,18 @@ def track_streaming_paths_for_sbom( added_path = False for asset in streaming_result.assets: if asset.path: - self.scanned_paths.append(_display_scan_path(asset.path)) + self.scanned_paths.append(asset.path) added_path = True if not added_path and fallback_path is not None and not os.path.exists(fallback_path): - self.scanned_paths.append(_display_scan_path(fallback_path)) + self.scanned_paths.append(fallback_path) def track_directory_paths_for_sbom(self, scan_result: ModelAuditResultModel) -> None: """Track completed directory scan assets, including an authoritative empty set.""" self.sbom_paths_resolved = True for asset in scan_result.assets: if asset.path: - self.scanned_paths.append(_display_scan_path(asset.path)) + self.scanned_paths.append(asset.path) def defer_temp_cleanup(self, temp_path: str | None, *, cache_enabled: bool, verbose: bool) -> None: """Track temporary artifacts for post-SBOM cleanup.""" @@ -1984,9 +1959,14 @@ def _track_huggingface_stream_acquisition( yielded_artifact = True yield streamed_item except Exception as exc: - if yielded_artifact: - raise _HuggingFaceStreamInterruptedError(str(exc)) from exc - raise _HuggingFaceAcquisitionError(str(exc)) from exc + message = str(exc) + error = ( + _HuggingFaceStreamInterruptedError(message) if yielded_artifact else _HuggingFaceAcquisitionError(message) + ) + cast(Any, error)._modelaudit_classification_text = _huggingface_classification_error( + getattr(exc, "_modelaudit_classification_text", message) + ) + raise error from exc finally: close_generator = getattr(file_generator, "close", None) if callable(close_generator): @@ -2044,7 +2024,7 @@ def _classify_huggingface_acquisition_error(error_msg: str) -> tuple[str, bool, def _huggingface_acquisition_source_key(path: str, requested_revision: str | None) -> str: - display_path = _display_scan_path(path) + display_path = path if requested_revision and not is_huggingface_file_url(path): return f"{display_path}@{requested_revision}" return display_path @@ -2056,12 +2036,18 @@ def _record_huggingface_acquisition_error( *, path: str, error_msg: str, + classification_error: object | None = None, scanned_artifact_count: int = 0, ) -> None: """Record a Hugging Face source failure while preserving completed scan evidence.""" requested_revision = _huggingface_requested_revision(path) source_key = _huggingface_acquisition_source_key(path, requested_revision) - reason, blocked, category = _classify_huggingface_acquisition_error(error_msg) + classification_message = ( + _escape_terminal_text(_huggingface_classification_error(classification_error)) + if classification_error is not None + else error_msg + ) + reason, blocked, category = _classify_huggingface_acquisition_error(classification_message) if scanned_artifact_count: artifact_label = "artifact was" if scanned_artifact_count == 1 else "artifacts were" issue_message = ( @@ -2101,7 +2087,17 @@ def _record_huggingface_acquisition_error( type="huggingface_acquisition_error", ) ) + identity_source = _huggingface_classification_url(path) + if requested_revision and not is_huggingface_file_url(path): + identity_source = f"{identity_source}@{requested_revision}" + preserve_finding_identity( + audit_result.issues[-1], + "huggingface_acquisition", + location=identity_source, + message=issue_message.replace(source_key, identity_source, 1), + ) audit_result.file_metadata[source_key] = FileMetadataModel( + source_identity={"producer": "huggingface_acquisition", "path": identity_source}, source="huggingface", source_url=source_key, acquisition_error=True, @@ -2828,26 +2824,66 @@ def _write_scan_sbom( if not sbom: return - from .integrations.sbom_generator import generate_sbom_pydantic + from .integrations.sbom_generator import ( + _cli_source_classification_path, + _source_identity_path, + generate_sbom_pydantic, + ) asset_paths = list( dict.fromkeys(asset.path for asset in audit_result.assets if asset.path and asset.type != "skipped") ) if asset_paths and (scan_and_delete or not path_state.sbom_paths_resolved): - paths_for_sbom = [_display_scan_path(path) for path in asset_paths] + paths_for_sbom = asset_paths elif path_state.sbom_paths_resolved: paths_for_sbom = path_state.scanned_paths else: - paths_for_sbom = ( - path_state.scanned_paths - if path_state.scanned_paths - else [_display_scan_path(path) for path in expanded_paths] - ) + paths_for_sbom = path_state.scanned_paths if path_state.scanned_paths else expanded_paths - sbom_text = generate_sbom_pydantic(paths_for_sbom, audit_result) + sbom_text = generate_sbom_pydantic( + paths_for_sbom, + audit_result, + _classification_paths={ + path: _cli_source_classification_path( + path if is_mlflow_uri(path) else _source_identity_path(path, audit_result.file_metadata.get(path)) + ) + for path in paths_for_sbom + }, + _reported_paths={path: _report_source_path(path) for path in paths_for_sbom}, + ) _write_output_text_file(sbom, sbom_text) +def _bound_report_finding_text(record: dict[str, Any], *, identity: bool = False) -> None: + """Retain historical text bounds before allocating source identifiers.""" + fields = ("message", "name", "type", "rule_code") + (() if identity else ("why", "recommendation")) + for field_name in fields: + value = record.get(field_name) + if isinstance(value, str): + record[field_name] = serialize_source_text(value) + details = record.get("details") + if isinstance(details, dict): + for field_name in ("evidence_fingerprint", "zip_entry_id", "zip_entry", "check_consolidation_key"): + value = details.get(field_name) + if isinstance(value, str): + details[field_name] = serialize_source_text(value) + + +def _serialize_scan_report(audit_result: ModelAuditResultModel, *, exclude_none: bool = False) -> dict[str, Any]: + result = audit_result.model_dump(mode="python", exclude_none=exclude_none) + for record in [*result["issues"], *result["checks"]]: + if finding_identity(record) is not record: + _bound_report_finding_text(record["finding_identity"]["fields"], identity=True) + _bound_report_finding_text(record) + sources = [ + *(asset.path for asset in audit_result.assets), + *(issue.location for issue in audit_result.issues if issue.location), + *(check.location for check in audit_result.checks if check.location), + *audit_result.file_metadata, + ] + return cast(dict[str, Any], serialize_source_value(result, identifier_key=_sarif_source_identifier_key(sources))) + + def _format_scan_output( audit_result: ModelAuditResultModel, expanded_paths: list[str], @@ -2860,14 +2896,14 @@ def _format_scan_output( if not verbose: audit_result.issues = [issue for issue in audit_result.issues if issue.severity != IssueSeverity.DEBUG] audit_result.checks = [check for check in audit_result.checks if check.severity != IssueSeverity.DEBUG] - redacted_result = redact_source_value(audit_result.model_dump(mode="python", exclude_none=True)) - return json.dumps(_JSON_VALUE_ADAPTER.dump_python(redacted_result, mode="json"), indent=2) + serialized_result = _serialize_scan_report(audit_result, exclude_none=True) + return json.dumps(_JSON_VALUE_ADAPTER.dump_python(serialized_result, mode="json"), indent=2) if output_format == "sarif": return format_sarif_output(audit_result, expanded_paths, verbose) - redacted_result = redact_source_value(audit_result.model_dump(mode="python")) - output_text = format_text_output(redacted_result if isinstance(redacted_result, dict) else {}, verbose) + serialized_result = _serialize_scan_report(audit_result) + output_text = format_text_output(serialized_result if isinstance(serialized_result, dict) else {}, verbose) previews = getattr(audit_result, "previews", None) if isinstance(previews, list) and previews: preview_text = _format_huggingface_dry_run_previews(previews, "text") @@ -3233,7 +3269,7 @@ def _scan_local_or_downloaded_path( elif os.path.isdir(actual_path): path_state.track_directory_paths_for_sbom(scan_results) else: - path_state.scanned_paths.append(_display_scan_path(actual_path)) + path_state.scanned_paths.append(actual_path) visible_issues = [ issue for issue in list(scan_results.issues) if verbose or issue.severity != IssueSeverity.DEBUG @@ -3295,7 +3331,7 @@ def _scan_local_or_downloaded_path( logger.error(f"Error during scan of {display_path}: {display_error}") click.echo(f"Error scanning {display_path}: {display_error}", err=True) path_state.mark_non_shard_error(audit_result) - path_state.scanned_paths.append(_display_scan_path(actual_path)) + path_state.scanned_paths.append(actual_path) if progress_tracker: progress_tracker.report_error(Exception(display_error)) @@ -3336,7 +3372,8 @@ def _resolve_scan_source_for_path( source_model_source=source_model_source, ) except Exception as exc: - error_msg = _display_error(exc, path) + raw_error = str(exc) + error_msg = _display_error(raw_error, path) logger.error(f"Failed to preview Hugging Face file {display_path}: {error_msg}") click.echo(f"Error previewing file from {display_path}: {error_msg}", err=True) _record_huggingface_acquisition_error( @@ -3344,6 +3381,7 @@ def _resolve_scan_source_for_path( path_state, path=path, error_msg=error_msg, + classification_error=getattr(exc, "_modelaudit_classification_text", raw_error), ) return None @@ -3401,7 +3439,8 @@ def _resolve_scan_source_for_path( elif runtime.show_styled_output: click.echo(style_text("❌ Download failed", fg="red", bold=True)) - error_msg = _display_error(exc, path) + raw_error = str(exc) + error_msg = _display_error(raw_error, path) logger.error(f"Failed to download file from {display_path}: {error_msg}") click.echo(f"Error downloading file from {display_path}: {error_msg}", err=True) _record_huggingface_acquisition_error( @@ -3409,6 +3448,7 @@ def _resolve_scan_source_for_path( path_state, path=path, error_msg=error_msg, + classification_error=getattr(exc, "_modelaudit_classification_text", raw_error), ) path_state.defer_temp_cleanup( temp_dir, @@ -3431,7 +3471,8 @@ def _resolve_scan_source_for_path( source_model_source=source_model_source, ) except Exception as exc: - error_msg = _display_error(exc, path) + raw_error = str(exc) + error_msg = _display_error(raw_error, path) logger.error(f"Failed to preview Hugging Face model {display_path}: {error_msg}") click.echo(f"Error previewing model from {display_path}: {error_msg}", err=True) _record_huggingface_acquisition_error( @@ -3439,6 +3480,7 @@ def _resolve_scan_source_for_path( path_state, path=path, error_msg=error_msg, + classification_error=getattr(exc, "_modelaudit_classification_text", raw_error), ) return None @@ -3674,7 +3716,8 @@ def _resolve_scan_source_for_path( if runtime.show_styled_output: click.echo(style_text("❌ Download/scan failed", fg="red", bold=True)) - error_msg = _display_error(exc, path) + raw_error = str(exc) + error_msg = _display_error(raw_error, path) if "insufficient disk space" in error_msg.lower(): logger.error(f"Disk space error for {display_path}: {error_msg}") click.echo(style_text(f"\n⚠️ {error_msg}", fg="yellow"), err=True) @@ -3695,6 +3738,7 @@ def _resolve_scan_source_for_path( path_state, path=path, error_msg=error_msg, + classification_error=getattr(exc, "_modelaudit_classification_text", raw_error), scanned_artifact_count=( streaming_result.files_scanned if streaming_result_aggregated and streaming_result is not None @@ -4844,7 +4888,7 @@ def scan_command( display_error = _display_error(exc, path) logger.error(f"Unexpected error processing {display_path}: {display_error}") click.echo(f"Unexpected error processing {display_path}: {display_error}", err=True) - path_state.scanned_paths.append(_display_scan_path(source_result.actual_path)) + path_state.scanned_paths.append(source_result.actual_path) path_state.mark_non_shard_error(audit_result) if progress_tracker: @@ -5919,31 +5963,6 @@ def _get_install_info() -> dict[str, Any]: return info -def _redact_proxy_url(proxy_url: str | None) -> str | None: - """Redact credentials from proxy URLs while preserving host/port for debugging. - - Proxy URLs often contain credentials (http://user:pass@host:port). - Since debug output is meant to be pasted in bug reports, we must redact - the credentials while keeping the scheme/host/port for troubleshooting. - """ - if not proxy_url: - return None - try: - from urllib.parse import urlsplit, urlunsplit - - parts = urlsplit(proxy_url) - if parts.username or parts.password: - # Rebuild URL without credentials - netloc = parts.hostname or "" - if parts.port: - netloc += f":{parts.port}" - return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment)) - except Exception: - # If parsing fails, return a safe indicator rather than the raw URL - return "" - return proxy_url - - def _get_env_info() -> dict[str, Any]: """Get environment variable information for debug output.""" from .telemetry import is_telemetry_enabled @@ -5954,8 +5973,8 @@ def _get_env_info() -> dict[str, Any]: "ciEnvironment": bool(os.getenv("CI")), "jfrogConfigured": bool(os.getenv("JFROG_API_TOKEN") or os.getenv("JFROG_ACCESS_TOKEN")), "mlflowConfigured": bool(os.getenv("MLFLOW_TRACKING_URI")), - "httpProxy": _redact_proxy_url(os.getenv("HTTP_PROXY") or os.getenv("http_proxy")), - "httpsProxy": _redact_proxy_url(os.getenv("HTTPS_PROXY") or os.getenv("https_proxy")), + "httpProxy": os.getenv("HTTP_PROXY") or os.getenv("http_proxy"), + "httpsProxy": os.getenv("HTTPS_PROXY") or os.getenv("https_proxy"), "noProxy": os.getenv("NO_PROXY") or os.getenv("no_proxy") or None, } @@ -6063,16 +6082,6 @@ def _get_scanner_info(verbose: bool = False) -> dict[str, Any]: return info -def _sanitize_debug_path(path: str) -> str: - """Sanitize filesystem paths for debug output.""" - home = str(Path.home()) - if path.startswith(home): - return "~" + path[len(home) :] - if os.path.isabs(path): - return "" - return path - - def _get_cache_info() -> dict[str, Any]: """Get cache information for debug output.""" try: @@ -6081,10 +6090,9 @@ def _get_cache_info() -> dict[str, Any]: cache_manager = get_cache_manager(enabled=True) stats = cache_manager.get_stats() - # Get cache directory path with ~ expansion for privacy cache_dir_path: str | None = None if cache_manager.cache is not None: - cache_dir_path = _sanitize_debug_path(str(cache_manager.cache.cache_dir)) + cache_dir_path = str(cache_manager.cache.cache_dir) return { "enabled": True, @@ -6105,17 +6113,15 @@ def _get_config_info() -> dict[str, Any]: # Shared config with promptfoo shared_config_dir = get_config_directory_path() shared_config_path = os.path.join(shared_config_dir, "promptfoo.yaml") - shared_display_path = _sanitize_debug_path(shared_config_path) # ModelAudit-specific config home = str(Path.home()) modelaudit_config_path = os.path.join(home, ".modelaudit", "user_config.json") - modelaudit_display_path = "~/.modelaudit/user_config.json" return { - "sharedConfigPath": shared_display_path, + "sharedConfigPath": shared_config_path, "sharedConfigExists": os.path.exists(shared_config_path), - "modelauditConfigPath": modelaudit_display_path, + "modelauditConfigPath": modelaudit_config_path, "modelauditConfigExists": os.path.exists(modelaudit_config_path), "userIdGenerated": bool(get_user_id()), } @@ -6146,7 +6152,7 @@ def _format_debug_output(debug_info: dict[str, Any], verbose: bool) -> str: lines.append(border) lines.append( style_text( - "Please include this output when reporting issues:", + "Inspect this output and remove sensitive values before sharing it in an issue report:", fg="yellow", ) ) @@ -6231,6 +6237,9 @@ def _format_debug_output(debug_info: dict[str, Any], verbose: bool) -> str: def debug(output_json: bool, verbose: bool) -> None: """Display debug information for troubleshooting. + Output includes raw configuration values. Inspect it and remove sensitive + values before sharing it in an issue report. + Outputs comprehensive diagnostic information useful for: \b diff --git a/modelaudit/core.py b/modelaudit/core.py index c16bce330..18502cdb7 100644 --- a/modelaudit/core.py +++ b/modelaudit/core.py @@ -28,6 +28,11 @@ def shared_source_sensitive_caches() -> Iterator[None]: import modelaudit.core_results as core_results +from modelaudit.integrations._sarif_identity_urls import ( + redact_cloud_error_for_display, + redact_stream_error_for_display, + redact_stream_url_for_display, +) from modelaudit.integrations.license_checker import ( LICENSE_FILES, check_commercial_use_warnings, @@ -154,6 +159,8 @@ def shared_source_sensitive_caches() -> Iterator[None]: should_defer_hash_for_file_backed_onnx, should_defer_hash_for_pytorch_read_limit, ) +from modelaudit.utils.helpers.evidence import format_terminal_text +from modelaudit.utils.helpers.finding_identity import preserve_finding_identity from modelaudit.utils.helpers.interrupt_handler import check_interrupted from modelaudit.utils.helpers.types import ( FilePath, @@ -177,19 +184,7 @@ def shared_source_sensitive_caches() -> Iterator[None]: _resolve_hf_cache_path, _trusted_hf_blobs_root, ) -from modelaudit.utils.sources.cloud_storage import ( - is_sensitive_credential_key, - is_stream_url, -) -from modelaudit.utils.sources.cloud_storage import ( - redact_cloud_error_for_display as _redact_cloud_error_for_display, -) -from modelaudit.utils.sources.cloud_storage import ( - redact_stream_error_for_display as _redact_stream_error_for_display, -) -from modelaudit.utils.sources.cloud_storage import ( - redact_stream_url_for_display as _redact_stream_url_for_display, -) +from modelaudit.utils.sources.cloud_storage import is_stream_url logger = logging.getLogger("modelaudit.core") @@ -949,87 +944,66 @@ def _make_trusted_stream_shard_root(path: FilePath) -> object: ) -def _redacted_stream_url_for_reporting(stream_url: str) -> str: - """Return a stream source identifier safe for persisted scan output.""" - return _redact_stream_url_for_display(stream_url) - - -def _redacted_scan_path_for_reporting(path: str) -> str: - if is_stream_url(path): - return f"stream://{_redacted_stream_url_for_reporting(path[9:])}" - return path - - -def _redacted_scan_error_for_reporting(error: object, path: str) -> str: - if is_stream_url(path): - return _redact_stream_error_for_display(error, path[9:]) - return str(error) - - -def _redact_stream_value_for_reporting(value: Any, stream_url: str, report_url: str) -> Any: +def _replace_report_path_value(value: Any, stream_url: str, report_url: str) -> Any: if isinstance(value, BaseModel): - return _redact_stream_value_for_reporting(value.model_dump(mode="python"), stream_url, report_url) + return _replace_report_path_value(value.model_dump(mode="python"), stream_url, report_url) if isinstance(value, AnyUrl): - return _redact_stream_value_for_reporting(str(value), stream_url, report_url) + return _replace_report_path_value(str(value), stream_url, report_url) if isinstance(value, os.PathLike): - return _redact_stream_value_for_reporting(os.fspath(value), stream_url, report_url) + return _replace_report_path_value(os.fspath(value), stream_url, report_url) if isinstance(value, str): - return _redact_cloud_error_for_display(value.replace(stream_url, report_url)) + return value.replace(stream_url, report_url) if isinstance(value, bytes): try: decoded = value.decode("utf-8") except UnicodeDecodeError: return b"" - return _redact_stream_value_for_reporting(decoded, stream_url, report_url).encode("utf-8") + return _replace_report_path_value(decoded, stream_url, report_url).encode("utf-8") if isinstance(value, bytearray): try: decoded = value.decode("utf-8") except UnicodeDecodeError: return bytearray(b"") - return bytearray(_redact_stream_value_for_reporting(decoded, stream_url, report_url), "utf-8") + return bytearray(_replace_report_path_value(decoded, stream_url, report_url), "utf-8") if isinstance(value, dict): - redacted_mapping: dict[Any, Any] = {} + rebased_mapping: dict[Any, Any] = {} for key, item in value.items(): - redacted_key = _redact_stream_value_for_reporting(key, stream_url, report_url) - redacted_mapping[redacted_key] = ( - "" - if is_sensitive_credential_key(key) - else _redact_stream_value_for_reporting(item, stream_url, report_url) - ) - return redacted_mapping + rebased_key = _replace_report_path_value(key, stream_url, report_url) + rebased_mapping[rebased_key] = _replace_report_path_value(item, stream_url, report_url) + return rebased_mapping if isinstance(value, list): - return [_redact_stream_value_for_reporting(item, stream_url, report_url) for item in value] + return [_replace_report_path_value(item, stream_url, report_url) for item in value] if isinstance(value, tuple): - return tuple(_redact_stream_value_for_reporting(item, stream_url, report_url) for item in value) + return tuple(_replace_report_path_value(item, stream_url, report_url) for item in value) if isinstance(value, set): - return {_redact_stream_value_for_reporting(item, stream_url, report_url) for item in value} + return {_replace_report_path_value(item, stream_url, report_url) for item in value} if isinstance(value, frozenset): - return frozenset(_redact_stream_value_for_reporting(item, stream_url, report_url) for item in value) + return frozenset(_replace_report_path_value(item, stream_url, report_url) for item in value) return value -def _redact_stream_record_for_reporting(record: Issue | Check, stream_url: str, report_url: str) -> None: +def _replace_record_report_path(record: Issue | Check, stream_url: str, report_url: str) -> None: for attr in ("location", "message", "why", "rule_code", "type", "name"): value = getattr(record, attr, None) if isinstance(value, str): - setattr(record, attr, _redact_stream_value_for_reporting(value, stream_url, report_url)) + setattr(record, attr, _replace_report_path_value(value, stream_url, report_url)) if record.details: - record.details = _redact_stream_value_for_reporting(record.details, stream_url, report_url) + record.details = _replace_report_path_value(record.details, stream_url, report_url) if record.model_extra: - redacted_extra = _redact_stream_value_for_reporting(record.model_extra, stream_url, report_url) + rebased_extra = _replace_report_path_value(record.model_extra, stream_url, report_url) record.model_extra.clear() - record.model_extra.update(redacted_extra) + record.model_extra.update(rebased_extra) -def _redact_stream_scan_result_for_reporting(scan_result: ScanResult, stream_url: str, report_url: str) -> None: - """Strip signed query material from scanner-owned records before aggregation.""" +def _replace_result_report_path(scan_result: ScanResult, stream_url: str, report_url: str) -> None: + """Replace descriptor-only paths in scanner records before aggregation.""" for issue in scan_result.issues: - _redact_stream_record_for_reporting(issue, stream_url, report_url) + _replace_record_report_path(issue, stream_url, report_url) for check in scan_result.checks: - _redact_stream_record_for_reporting(check, stream_url, report_url) + _replace_record_report_path(check, stream_url, report_url) if scan_result.metadata: - scan_result.metadata = _redact_stream_value_for_reporting(scan_result.metadata, stream_url, report_url) + scan_result.metadata = _replace_report_path_value(scan_result.metadata, stream_url, report_url) scan_result._refresh_metadata_dependent_state() @@ -1062,6 +1036,29 @@ def _rebase_bound_directory_owner_value_for_reporting(value: Any, report_root: P return value +def _preserve_scan_result_identity(scan_result: ScanResult, producer: str, source: str, identity_source: str) -> None: + """Retain producer normalization before aggregation consumes identity fields.""" + + def normalized(value: str) -> str: + return redact_cloud_error_for_display(value.replace(source, identity_source)) + + records: list[Issue | Check] = [*scan_result.issues, *scan_result.checks] + for record in records: + fields: dict[str, Any] = { + name: normalized(value) + for name in ("message", "location", "type", "rule_code", "name") + if isinstance(value := getattr(record, name, None), str) + } + details = { + name: normalized(value) + for name in ("evidence_fingerprint", "zip_entry_id", "zip_entry", "check_consolidation_key") + if isinstance(value := record.details.get(name), str) + } + if details: + fields["details"] = details + preserve_finding_identity(record, producer, **fields) + + def _normalize_directory_owner_scan_result_for_reporting( scan_result: ScanResult, owner_scan_path: str, @@ -1069,7 +1066,8 @@ def _normalize_directory_owner_scan_result_for_reporting( ) -> None: """Rewrite descriptor-only owner scan paths before aggregate reporting.""" if owner_scan_path != os.curdir: - _redact_stream_scan_result_for_reporting(scan_result, owner_scan_path, report_path) + _replace_result_report_path(scan_result, owner_scan_path, report_path) + _preserve_scan_result_identity(scan_result, "directory_owner", report_path, report_path) return report_root = Path(report_path) @@ -3553,7 +3551,7 @@ def scan_model_directory_or_file( if is_stream_url(path): # Extract the actual URL stream_url = path[9:] # Remove "stream://" prefix - report_url = _redacted_stream_url_for_reporting(stream_url) + report_url = stream_url if progress_callback: progress_callback(f"Streaming analysis: {report_url}", 0.0) @@ -3571,7 +3569,10 @@ def scan_model_directory_or_file( else: scan_result, analysis_complete = stream_analyze_file(stream_url, scanner) if scan_result: - _redact_stream_scan_result_for_reporting(scan_result, stream_url, report_url) + _replace_result_report_path(scan_result, stream_url, stream_url) + identity_source = redact_stream_url_for_display(stream_url) + _preserve_scan_result_identity(scan_result, "stream", stream_url, identity_source) + scan_result.metadata["source_identity"] = {"producer": "stream", "path": identity_source} if not analysis_complete: _mark_inconclusive_scan_outcome(scan_result, "streaming_analysis_incomplete") results.files_scanned += 1 @@ -3590,6 +3591,9 @@ def scan_model_directory_or_file( location=report_url, details={"analysis_complete": False}, ) + preserve_finding_identity( + results.issues[-1], "stream", location=redact_stream_url_for_display(stream_url) + ) else: raise ValueError(f"Streaming analysis failed for {report_url}") else: @@ -3688,10 +3692,7 @@ def merge_directory_owner_result(owner_result: ScanResult, *, dispatched: bool) directory_owner_result.add_check( name="Directory Owner Scan", passed=False, - message=( - "Unable to complete logical model-directory analysis: " - f"{_redacted_scan_error_for_reporting(error, path)}" - ), + message=(f"Unable to complete logical model-directory analysis: {error!s}"), severity=IssueSeverity.INFO, location=path, details={ @@ -4893,10 +4894,7 @@ def owner_source_covered_by_child(source: str) -> bool: directory_owner_result.add_check( name="Directory Owner Scan", passed=False, - message=( - "Unable to complete logical model-directory analysis: " - f"{_redacted_scan_error_for_reporting(error, path)}" - ), + message=(f"Unable to complete logical model-directory analysis: {error!s}"), severity=IssueSeverity.INFO, location=path, details={ @@ -5745,10 +5743,10 @@ def owner_source_covered_by_child(source: str) -> bool: results, "Scan interrupted by user", severity=IssueSeverity.INFO.value, details={"interrupted": True} ) except Exception as e: - report_path = _redacted_scan_path_for_reporting(path) - report_error = _redacted_scan_error_for_reporting(e, path) + report_path = path + report_error = str(e) if is_stream_url(path): - logger.error(f"Error during scan: {report_error}") + logger.error(f"Error during scan: {format_terminal_text(report_error)}") else: logger.exception(f"Error during scan: {report_error}") scan_metadata["success"] = False @@ -5759,6 +5757,17 @@ def owner_source_covered_by_child(source: str) -> bool: severity=IssueSeverity.INFO.value, details={"exception_type": type(e).__name__}, ) + if is_stream_url(path): + from .models import FileMetadataModel + + results.file_metadata[report_path] = FileMetadataModel( + source_identity={"producer": "stream", "path": f"stream://{redact_stream_url_for_display(path[9:])}"} + ) + preserve_finding_identity( + results.issues[-1], + "stream", + message=f"Error during scan: {redact_stream_error_for_display(e, path[9:])}", + ) _add_error_asset_to_results(results, report_path) finally: pickle_source_snapshot_stack.close() diff --git a/modelaudit/core_results.py b/modelaudit/core_results.py index b73410e6e..38e28d286 100644 --- a/modelaudit/core_results.py +++ b/modelaudit/core_results.py @@ -22,6 +22,7 @@ ) from modelaudit.telemetry import record_issue_found from modelaudit.utils.helpers.assets import asset_from_scan_result +from modelaudit.utils.helpers.finding_identity import finding_identity, preserve_finding_identity logger = logging.getLogger("modelaudit.core.results") @@ -699,10 +700,11 @@ def _group_checks_by_asset(checks_list: list[Any]) -> dict[tuple[str, str], list logger.warning(f"Invalid check format at index {i}, skipping: {type(check)}") continue - check_name = check.get("name", "Unknown Check") - location = check.get("location", "") + identity = finding_identity(check) + check_name = identity.get("name", "Unknown Check") + location = identity.get("location", "") primary_asset = _extract_primary_asset_from_location(location) - details = check.get("details") + details = identity.get("details") zip_entry_id = details.get("zip_entry_id") if isinstance(details, dict) else None zip_entry = details.get("zip_entry") if isinstance(details, dict) else None @@ -848,15 +850,33 @@ def consolidate_checks(results: ModelAuditResultModel) -> None: else: consolidated_status = "skipped" + identity_checks = [finding_identity(check) for check in group_checks] + identity_message = _create_consolidated_message(check_name, identity_checks, consolidated_status, failed_count) + raw_check_name = group_checks[0].get("name", check_name) + message = next( + ( + check["message"] + for check, identity in zip(group_checks, identity_checks, strict=True) + if identity.get("message") == identity_message + ), + _create_consolidated_message(raw_check_name, identity_checks, consolidated_status, failed_count), + ) consolidated_check = { - "name": check_name, + "name": raw_check_name, "status": consolidated_status, - "message": _create_consolidated_message(check_name, group_checks, consolidated_status, failed_count), + "message": message, "location": group_checks[0].get("location", primary_asset), "details": _collect_consolidated_details(group_checks), "timestamp": _get_consolidated_timestamp(group_checks), } + preserve_finding_identity( + consolidated_check, + "check_consolidation", + name=check_name, + message=identity_message, + location=identity_checks[0].get("location", primary_asset), + ) consolidated_severity, consolidated_why = _extract_failure_context(group_checks) if consolidated_severity: consolidated_check["severity"] = consolidated_severity diff --git a/modelaudit/detectors/jit_script.py b/modelaudit/detectors/jit_script.py index f917895c3..280f04b49 100644 --- a/modelaudit/detectors/jit_script.py +++ b/modelaudit/detectors/jit_script.py @@ -19,6 +19,8 @@ from collections.abc import Callable, Collection, Iterator, Mapping, Sequence from typing import TYPE_CHECKING, Any, TypeVar +from modelaudit.utils.helpers.evidence import format_evidence_string + if TYPE_CHECKING: from modelaudit.models import JITScriptFinding @@ -53,13 +55,6 @@ def create_jit_finding(**kwargs: Any) -> "JITScriptFinding": return JITScriptFinding(**kwargs) -def _redact_code_evidence_snippet(code: str, max_chars: int = 200) -> str: - """Redact credentials from detector code evidence before serializing it.""" - from modelaudit.scanners._evidence_redaction import redact_evidence_string - - return redact_evidence_string(code, max_chars=max_chars) - - # Dangerous TorchScript operations that can execute arbitrary code DANGEROUS_TORCH_OPS = [ # System operations @@ -22717,7 +22712,7 @@ def _extract_and_check_python_code( recommendation=f"Remove {dangerous_import} import - it can be used maliciously", confidence=0.9, framework=framework, - code_snippet=_redact_code_evidence_snippet(code_str), + code_snippet=format_evidence_string(code_str, max_chars=200), type="dangerous_import", operation=None, builtin=None, @@ -22742,7 +22737,7 @@ def _extract_and_check_python_code( recommendation=f"Remove {builtin} usage - it can execute arbitrary code", confidence=0.9, framework=framework, - code_snippet=_redact_code_evidence_snippet(code_str), + code_snippet=format_evidence_string(code_str, max_chars=200), type="dangerous_builtin", operation=None, builtin=builtin, @@ -22864,7 +22859,7 @@ def _extract_and_check_python_code( recommendation=f"Remove {builtin} usage - it can execute arbitrary code", confidence=0.9, framework=framework, - code_snippet=_redact_code_evidence_snippet(code_str), + code_snippet=format_evidence_string(code_str, max_chars=200), type="dangerous_builtin", operation=None, builtin=builtin, diff --git a/modelaudit/detectors/secrets.py b/modelaudit/detectors/secrets.py index ddd3c2848..1f02f362a 100644 --- a/modelaudit/detectors/secrets.py +++ b/modelaudit/detectors/secrets.py @@ -13,6 +13,8 @@ import re from typing import Any +from modelaudit.utils.helpers.evidence import format_evidence_string + logger: logging.Logger = logging.getLogger(__name__) BASIC_AUTH_SECRET_TYPE = "Basic Auth Credentials" @@ -1454,7 +1456,7 @@ def _record_basic_auth_finding( "length": len(token), "confidence": round(confidence, 2), "pattern": pattern.pattern[:50] + "..." if len(pattern.pattern) > 50 else pattern.pattern, - "redacted_value": "Basic ", + "redacted_value": format_evidence_string(matched_text), "message": f"{BASIC_AUTH_SECRET_TYPE} detected (confidence: {confidence:.0%})", "context": f"{safe_context} pos:{position}" if safe_context else f"pos:{position}", "recommendation": f"Remove {BASIC_AUTH_SECRET_TYPE} from model data immediately", @@ -1768,9 +1770,6 @@ def scan_text(self, text: str, context: str = "", is_binary_source: bool = False else: severity = "INFO" - # Redact the secret for safe reporting - redacted = secret_text[:4] + "***" + secret_text[-4:] if len(secret_text) > 10 else "***REDACTED***" - if not self._record_finding( findings, { @@ -1781,7 +1780,7 @@ def scan_text(self, text: str, context: str = "", is_binary_source: bool = False "length": len(secret_text), "confidence": round(confidence, 2), "pattern": pattern.pattern[:50] + "..." if len(pattern.pattern) > 50 else pattern.pattern, - "redacted_value": redacted, + "redacted_value": format_evidence_string(secret_text), "message": f"{description} detected (confidence: {confidence:.0%})", "context": f"{safe_context} pos:{position}" if safe_context else f"pos:{position}", "recommendation": f"Remove {description} from model data immediately" diff --git a/modelaudit/integrations/_mlflow_identity.py b/modelaudit/integrations/_mlflow_identity.py new file mode 100644 index 000000000..99c986460 --- /dev/null +++ b/modelaudit/integrations/_mlflow_identity.py @@ -0,0 +1,172 @@ +"""Historical bounded source identity for MLflow acquisition findings.""" + +import re +from urllib.parse import unquote + +from modelaudit.detectors.network_comm import _redact_urls_in_text +from modelaudit.integrations._sarif_identity_urls import redact_cloud_error_for_display, redact_url_for_display +from modelaudit.scanners._evidence_redaction import ( + MAX_PERCENT_DECODE_PASSES, + MAX_REDACTION_VALUE_DEPTH, + SENSITIVE_CONTAINER_KEY, + redact_evidence_string, +) + +_MAX_MLFLOW_ERROR_DISPLAY_CHARS = 512 +_MLFLOW_SENSITIVE_KEY = rf"(?:{SENSITIVE_CONTAINER_KEY}|credentials?|jwt|session)" +_MLFLOW_SENSITIVE_ASSIGNMENT_RE = re.compile( + r"(?ix)" + r"(?P(?\"(?:\\.|[^\"\\])*\"|'(?:\\.|[^'\\])*'|(?:(?:bearer|basic|token)\s+)?[^\s,;&}\]]+)" +) +_MLFLOW_BRACKETED_SENSITIVE_ASSIGNMENT_RE = re.compile( + r"(?ix)" + r"(?P(?\"(?:\\.|[^\"\\])*\"|'(?:\\.|[^'\\])*'|(?:(?:bearer|basic|token)\s+)?[^\s,;&}\]]+)" +) +_MLFLOW_SENSITIVE_CONTAINER_PREFIX_RE = re.compile( + r"(?ix)" + r"(?P(?[({\[])", +) +_MLFLOW_PROTOCOL_RELATIVE_URL_RE = re.compile( + r"(?i)(?:(?:[\\/]|%(?:25)*(?:2f|5c)){2,})[^\s\"'<>]+", +) +_MLFLOW_BENIGN_AUTH_CONTEXT_RE = re.compile( + r"(?i)\b(?:bearer|basic|token)(?=\s+(?:authentication|endpoint|refresh|service)\b)", +) + + +def _redact_mlflow_error_for_display(error: object) -> str: + def _replace_sensitive_value(match: re.Match[str]) -> str: + value = match.group("value") + quote = value[0] if value[:1] in {'"', "'"} else "" + return f"{match.group('prefix')}{quote}{quote}" + + def _redact_protocol_relative_url(match: re.Match[str]) -> str: + candidate = match.group(0) + decoded = candidate + for _ in range(MAX_PERCENT_DECODE_PASSES): + next_decoded = unquote(decoded) + if next_decoded == decoded: + break + decoded = next_decoded + + normalized = decoded.replace("\\/", "/").replace("\\", "/") + if len(normalized) - len(normalized.lstrip("/")) < 2: + return candidate + normalized = f"//{normalized.lstrip('/')}" + authority = normalized[2:].split("/", 1)[0].split("?", 1)[0].split("#", 1)[0] + if "@" not in authority: + return candidate + + safe_url = redact_url_for_display(f"https:{normalized}") + return safe_url.removeprefix("https:") + + def _redact_sensitive_containers(text: str) -> str: + parts: list[str] = [] + cursor = 0 + closing_delimiters = {"(": ")", "[": "]", "{": "}"} + + while match := _MLFLOW_SENSITIVE_CONTAINER_PREFIX_RE.search(text, cursor): + parts.append(text[cursor : match.start()]) + parts.append(f"{match.group('prefix')}") + stack = [closing_delimiters[match.group("open")]] + quote: str | None = None + escaped = False + index = match.end() + + while index < len(text) and stack: + character = text[index] + if quote is not None: + if escaped: + escaped = False + elif character == "\\": + escaped = True + elif character == quote: + quote = None + elif character in {'"', "'"}: + quote = character + elif character in closing_delimiters: + if len(stack) >= MAX_REDACTION_VALUE_DEPTH: + index = len(text) + break + stack.append(closing_delimiters[character]) + elif character == stack[-1]: + stack.pop() + index += 1 + + if stack: + cursor = len(text) + break + cursor = index + + parts.append(text[cursor:]) + return "".join(parts) + + redacted = _MLFLOW_PROTOCOL_RELATIVE_URL_RE.sub(_redact_protocol_relative_url, str(error)) + redacted = _redact_sensitive_containers(redacted) + redacted = _MLFLOW_BRACKETED_SENSITIVE_ASSIGNMENT_RE.sub(_replace_sensitive_value, redacted) + redacted = _MLFLOW_SENSITIVE_ASSIGNMENT_RE.sub(_replace_sensitive_value, redacted) + contains_url = bool( + re.search(r"(?i)(?:\b[a-z][a-z0-9+.-]*://|\bmodels:/)", redacted) + or _MLFLOW_PROTOCOL_RELATIVE_URL_RE.search(redacted) + ) + if contains_url: + redacted = redact_cloud_error_for_display(_redact_urls_in_text(redacted)) + redacted = _MLFLOW_PROTOCOL_RELATIVE_URL_RE.sub(_redact_protocol_relative_url, redacted) + redacted = _MLFLOW_BRACKETED_SENSITIVE_ASSIGNMENT_RE.sub(_replace_sensitive_value, redacted) + redacted = _MLFLOW_SENSITIVE_ASSIGNMENT_RE.sub(_replace_sensitive_value, redacted) + + benign_auth_contexts: list[tuple[str, str]] = [] + + def _protect_benign_auth_context(match: re.Match[str]) -> str: + placeholder = f"MODELAUDITMLFLOWSAFECONTEXT{len(benign_auth_contexts)}" + benign_auth_contexts.append((placeholder, match.group(0))) + return placeholder + + redacted = _MLFLOW_BENIGN_AUTH_CONTEXT_RE.sub(_protect_benign_auth_context, redacted) + if contains_url: + redacted = redact_evidence_string(redacted, max_chars=None) + else: + redacted = "&".join(redact_evidence_string(part, max_chars=None) for part in redacted.split("&")) + for placeholder, original in benign_auth_contexts: + redacted = redacted.replace(placeholder, original) + + if len(redacted) <= _MAX_MLFLOW_ERROR_DISPLAY_CHARS: + return redacted + return f"{redacted[: _MAX_MLFLOW_ERROR_DISPLAY_CHARS - 3]}..." + + +def _mlflow_text_requires_specialized_redaction(text: str) -> bool: + if "models:/" in text.lower(): + return True + if _MLFLOW_BRACKETED_SENSITIVE_ASSIGNMENT_RE.search(text) or _MLFLOW_SENSITIVE_CONTAINER_PREFIX_RE.search(text): + return True + for match in _MLFLOW_PROTOCOL_RELATIVE_URL_RE.finditer(text): + candidate = match.group(0) + if candidate.startswith("//") and re.search(r"(?i)[a-z][a-z0-9+.-]*:$", text[: match.start()]): + continue + decoded = candidate + for _ in range(MAX_PERCENT_DECODE_PASSES): + next_decoded = unquote(decoded) + if next_decoded == decoded: + break + decoded = next_decoded + normalized = decoded.replace("\\/", "/").replace("\\", "/") + authority = normalized.lstrip("/").split("/", 1)[0].split("?", 1)[0].split("#", 1)[0] + if "@" in authority: + return True + return False + + +def mlflow_source_identity(value: str) -> str: + specialized = ( + _redact_mlflow_error_for_display(value) if _mlflow_text_requires_specialized_redaction(value) else value + ) + return redact_evidence_string(specialized, max_chars=_MAX_MLFLOW_ERROR_DISPLAY_CHARS) diff --git a/modelaudit/integrations/source_redaction.py b/modelaudit/integrations/_sarif_identity.py similarity index 80% rename from modelaudit/integrations/source_redaction.py rename to modelaudit/integrations/_sarif_identity.py index afaeab949..367d6560b 100644 --- a/modelaudit/integrations/source_redaction.py +++ b/modelaudit/integrations/_sarif_identity.py @@ -1,39 +1,48 @@ -"""Credential-safe source identifier redaction for exported reports.""" +"""Historical normalization for SARIF derived identities only. + +SBOM component classification also uses normalized identifiers. Raw report +messages, paths, and properties do not use this normalization. Keep +these inputs stable so existing finding fingerprints and derived rule grouping +survive credential rotation and remain compatible with previously exported SARIF. +""" import os import re -from typing import Any -from urllib.parse import parse_qsl, unquote, urlencode, urlsplit, urlunsplit - -from pydantic import AnyUrl, BaseModel +from urllib.parse import unquote, urlsplit, urlunsplit -from modelaudit.utils.sources.cloud_storage import ( +from modelaudit.integrations._sarif_identity_urls import ( _normalize_percent_encoded_url_authority_for_display as _normalize_percent_encoded_url_authority_for_display, ) -from modelaudit.utils.sources.cloud_storage import ( +from modelaudit.integrations._sarif_identity_urls import ( _normalize_percent_encoded_url_delimiters_for_display as _normalize_percent_encoded_url_delimiters_for_display, ) -from modelaudit.utils.sources.cloud_storage import is_sensitive_credential_key, is_stream_url -from modelaudit.utils.sources.cloud_storage import ( +from modelaudit.integrations._sarif_identity_urls import is_sensitive_credential_key, is_stream_url +from modelaudit.integrations._sarif_identity_urls import ( normalize_escaped_url_delimiters_for_display as _normalize_escaped_url_delimiters_for_display, ) -from modelaudit.utils.sources.cloud_storage import redact_cloud_error_for_display as _redact_cloud_error_for_display -from modelaudit.utils.sources.cloud_storage import redact_stream_url_for_display as _redact_stream_url_for_display -from modelaudit.utils.sources.cloud_storage import redact_url_for_display as _redact_url_for_display +from modelaudit.integrations._sarif_identity_urls import ( + redact_cloud_error_for_display as _redact_cloud_error_for_display, +) +from modelaudit.integrations._sarif_identity_urls import redact_stream_url_for_display as _redact_stream_url_for_display +from modelaudit.integrations._sarif_identity_urls import redact_url_for_display as _redact_url_for_display _URL_TEXT_CHARACTER = r'(?:[^\s"\'<>]||)' + _URL_TOKEN_RE = re.compile( rf"(?(?:(?P[a-z][a-z0-9+.-]*):/{1,2}|//)?)" r"(?P[^/\s?#@]+(?:@|%(?:25)*40))" r"(?P[^/\s?#@]+)(?P.*)$", re.IGNORECASE, ) + _USERINFO_TOKEN_RE = re.compile( r"(?" r"(?:(?:[a-z][a-z0-9+.-]*):/{1,2}|//)?" @@ -42,6 +51,7 @@ r")", re.IGNORECASE, ) + _SCHEMELESS_SUFFIX_TOKEN_RE = re.compile( rf"(?" rf"(?:(?:[a-z]:[\\/]|/|\.\.?/)?(?:[^\s\"'<>/?#]+/)+)" @@ -49,6 +59,7 @@ rf")", re.IGNORECASE, ) + _SCHEMELESS_ENCODED_SUFFIX_TOKEN_RE = re.compile( rf"(?" rf"(?:(?:[a-z]:[\\/]|/|\.\.?/)?(?:[^\s\"'<>/?#]+/)+)" @@ -56,6 +67,7 @@ rf")", re.IGNORECASE, ) + _BARE_SUFFIX_TOKEN_RE = re.compile( rf"(?" rf"[0-9A-Za-z._~-]+\.[A-Za-z][0-9A-Za-z]{{0,15}}" @@ -63,6 +75,7 @@ rf")", re.IGNORECASE, ) + _BARE_ENCODED_SUFFIX_TOKEN_RE = re.compile( rf"(?" rf"[0-9A-Za-z._~-]+\.[A-Za-z][0-9A-Za-z]{{0,15}}" @@ -70,6 +83,7 @@ rf")", re.IGNORECASE, ) + _EMAIL_SUFFIX_TOKEN_RE = re.compile( rf"(?" rf"[0-9A-Za-z._%+-]+@[0-9A-Za-z.-]+\.[A-Za-z]{{2,}}" @@ -77,24 +91,38 @@ rf")", re.IGNORECASE, ) + _WINDOWS_DRIVE_PATH_RE = re.compile(r"^[a-z]:[\\/]", re.IGNORECASE) + _ENCODED_ASSIGNMENT_SEPARATOR_RE = re.compile(r"%(?:25)*(?:3a|3d)", re.IGNORECASE) + _ENCODED_MAJOR_SUFFIX_RE = re.compile(r"%(?:25)*(?:3f|23|3b)", re.IGNORECASE) + _ENCODED_FILENAME_SUFFIX_RE = re.compile(r"^[0-9A-Za-z._~-]+\.[0-9A-Za-z]{1,16}$") + _ENCODED_AT_RE = re.compile(r"%(?:25)*40", re.IGNORECASE) + _EXPORT_KEY_TOKEN = r"(?:[0-9A-Za-z_%.-]|\\(?:u[0-9A-Fa-f]{4}|x[0-9A-Fa-f]{2}))+" + _EXPORT_BRACKET_KEY = rf"\[\s*(?:{_EXPORT_KEY_TOKEN}|\"{_EXPORT_KEY_TOKEN}\"|'{_EXPORT_KEY_TOKEN}')?\s*\]" + _EXPORT_QUOTED_VALUE = r"""(?:"(?:\\.|[^"\\])*"|'(?:\\.|[^'\\])*')""" + _ESCAPED_KEY_CHARACTER_RE = re.compile( r"\\(?:u(?P[0-9A-Fa-f]{4})|x(?P[0-9A-Fa-f]{2}))", re.IGNORECASE, ) -_MAX_REDACTION_DEPTH = 32 + _MAX_SOURCE_TEXT_CHARS = 256 * 1024 + _MAX_PROVENANCE_QUERY_CHARS = 4096 + _MAX_PROVENANCE_PARAMS = 16 + _SAFE_PROVENANCE_QUERY_KEYS = frozenset({"branch", "ref", "revision", "tag", "version"}) + _SAFE_PROVENANCE_VALUE_RE = re.compile(r"^[0-9A-Za-z._~:+/-]{1,128}$") + _EXPORT_ASSIGNMENT_RE = re.compile( r"(?\\?)(?P[\"'])" @@ -103,6 +131,7 @@ r"(?P\s*(?::|(?])=(?!=)|<\\?)(?P[\"'])" @@ -111,23 +140,28 @@ r"(?P\s*={2,}\s*)", re.IGNORECASE, ) + _EXPORT_EQUALS_KEY_RE = re.compile( r"(?[0-9A-Za-z_%.-]+)(?P\s*=\s*)", re.IGNORECASE, ) + _EXPORT_HEADER_KEY_RE = re.compile( r"(?[\"']?)(?P[0-9A-Za-z_%.-]+)(?P=quote)(?P\s*:\s*)", re.IGNORECASE, ) + _EXPORT_ENCODED_SEPARATOR_RE = re.compile( r"(?[0-9A-Za-z_%.-]+?)" r"(?P%(?:25)*(?P3a|3d)(?:%(?:25)*20)*)", re.IGNORECASE, ) + _EXPORT_OPTION_RE = re.compile( rf"(?--(?P[0-9A-Za-z_%.-]+)\s+)(?P{_EXPORT_QUOTED_VALUE}|[^\s,;]+)", re.IGNORECASE, ) + _EXPORT_AUTHORIZATION_RE = re.compile( r"(?(?P(?:proxy[_.-]?)?authorization)\s+" @@ -136,8 +170,11 @@ rf"(?P{_EXPORT_QUOTED_VALUE}|[^\s,;]+)", re.IGNORECASE, ) + _EXPORT_VALUE_BOUNDARY_RE = re.compile(r"[\r\n,;)}\]]|\s+(?=--[0-9A-Za-z])") + _MAX_CREDENTIAL_KEY_DECODE_PASSES = 4 + _EXPORT_CREDENTIAL_KEY_ALIASES = frozenset( { "dbpassword", @@ -151,6 +188,7 @@ "sessiontoken", } ) + _EXPORT_CREDENTIAL_KEY_TOKENS = frozenset( { "auth", @@ -167,6 +205,7 @@ "token", } ) + _EXPORT_CREDENTIAL_KEY_NEAR_MATCHES = frozenset( { "accesstokencount", @@ -189,6 +228,7 @@ "tokenizer", } ) + _EXPORT_SAFE_METADATA_KEY_SUFFIXES = ( "authmethod", "authenticationmethod", @@ -198,6 +238,7 @@ "signaturealgorithm", "tokencount", ) + _CREDENTIAL_SHAPED_PROVENANCE_VALUE_RE = re.compile( r"(?:gh[pousr]_[0-9A-Za-z_]{20,}|sk-[0-9A-Za-z_-]{12,}|AKIA[0-9A-Z]{16}|" r"eyJ[0-9A-Za-z_-]{8,}\.[0-9A-Za-z_-]{8,}\.[0-9A-Za-z_-]{8,})" @@ -289,11 +330,6 @@ def redact_source_text(text: str) -> str: return _redact_source_text(text, preserve_redacted_assignments=False) -def _redact_prevalidated_source_text(text: str) -> str: - """Redact source identifiers after a domain sanitizer validated markers.""" - return _redact_source_text(text, preserve_redacted_assignments=True) - - def _redact_source_text(text: str, *, preserve_redacted_assignments: bool) -> str: if len(text) > _MAX_SOURCE_TEXT_CHARS: return "" @@ -365,37 +401,6 @@ def _redact_url_adjacent_assignments(text: str) -> str: return f"{text[:first_assignment]}{_redact_export_alias_assignments(text[first_assignment:])}" -def redact_source_reference(source: str) -> str: - """Return a credential-safe source reference with bounded provenance context.""" - safe_identifier = redact_source_identifier(source) - if safe_identifier == source: - return safe_identifier - normalized_source = _normalize_percent_encoded_url_delimiters_for_display( - _normalize_escaped_url_delimiters_for_display(source) - ) - try: - parts = urlsplit(normalized_source) - except Exception: - return safe_identifier - - safe_params: list[tuple[str, str]] = [] - for raw_params, key_prefix in ((parts.query, ""), (parts.fragment, "fragment-")): - if not raw_params or len(raw_params) > _MAX_PROVENANCE_QUERY_CHARS: - continue - for key, value in parse_qsl(raw_params, keep_blank_values=True)[:_MAX_PROVENANCE_PARAMS]: - normalized_key = key.casefold() - if ( - normalized_key in _SAFE_PROVENANCE_QUERY_KEYS - and not _is_sensitive_export_key(normalized_key) - and _SAFE_PROVENANCE_VALUE_RE.fullmatch(value) - and not _looks_like_credential_value(value) - ): - safe_params.append((f"{key_prefix}{normalized_key}", value)) - if not safe_params: - return safe_identifier - return f"{safe_identifier}?{urlencode(sorted(safe_params))}" - - def _has_safe_schemeless_provenance_suffix(source: str) -> bool: """Preserve bounded, explicitly non-sensitive assignments in local-looking names.""" normalized_source = _normalize_percent_encoded_url_delimiters_for_display( @@ -425,136 +430,6 @@ def _has_safe_schemeless_provenance_suffix(source: str) -> bool: return True -def redact_source_value(value: Any) -> Any: - """Recursively redact exported values that may contain source identifiers.""" - return _redact_source_value( - value, - seen=set(), - depth=0, - preserve_redacted_assignments=False, - ) - - -def redact_prevalidated_source_value(value: Any) -> Any: - """Redact source identifiers after a domain sanitizer validated markers.""" - return _redact_source_value( - value, - seen=set(), - depth=0, - preserve_redacted_assignments=True, - ) - - -def _redact_source_value( - value: Any, - *, - seen: set[int], - depth: int, - preserve_redacted_assignments: bool, -) -> Any: - if depth > _MAX_REDACTION_DEPTH: - return "" - if isinstance(value, BaseModel): - return _redact_source_value( - value.model_dump(mode="python"), - seen=seen, - depth=depth + 1, - preserve_redacted_assignments=preserve_redacted_assignments, - ) - if isinstance(value, AnyUrl): - return redact_source_text(str(value)) - if isinstance(value, str): - if preserve_redacted_assignments: - return _redact_prevalidated_source_text(value) - return redact_source_text(value) - if isinstance(value, (bytes, bytearray)): - try: - decoded = bytes(value).decode("utf-8") - if preserve_redacted_assignments: - return _redact_prevalidated_source_text(decoded) - return redact_source_text(decoded) - except UnicodeDecodeError: - return "" - if isinstance(value, dict): - if id(value) in seen: - return "" - seen.add(id(value)) - try: - redacted_mapping: dict[Any, Any] = {} - next_key_occurrences: dict[str, int] = {} - for key, item in value.items(): - redacted_key = _unique_redacted_mapping_key( - _redact_mapping_key(key), - redacted_mapping, - next_occurrences=next_key_occurrences, - ) - redacted_mapping[redacted_key] = ( - "" - if _mapping_key_requires_redaction(key) - else _redact_source_value( - item, - seen=seen, - depth=depth + 1, - preserve_redacted_assignments=preserve_redacted_assignments, - ) - ) - return redacted_mapping - finally: - seen.remove(id(value)) - if isinstance(value, list): - if id(value) in seen: - return "" - seen.add(id(value)) - try: - return [ - _redact_source_value( - item, - seen=seen, - depth=depth + 1, - preserve_redacted_assignments=preserve_redacted_assignments, - ) - for item in value - ] - finally: - seen.remove(id(value)) - if isinstance(value, tuple): - if id(value) in seen: - return "" - seen.add(id(value)) - try: - return tuple( - _redact_source_value( - item, - seen=seen, - depth=depth + 1, - preserve_redacted_assignments=preserve_redacted_assignments, - ) - for item in value - ) - finally: - seen.remove(id(value)) - if isinstance(value, (set, frozenset)): - if id(value) in seen: - return "" - seen.add(id(value)) - try: - return sorted( - ( - _redact_source_value( - item, - seen=seen, - depth=depth + 1, - preserve_redacted_assignments=preserve_redacted_assignments, - ) - for item in value - ), - key=repr, - ) - finally: - seen.remove(id(value)) - return value - - def _redact_userinfo_identifier(source: str) -> str | None: match = _USERINFO_IDENTIFIER_RE.match(source) if match is None: @@ -737,34 +612,6 @@ def _redact_local_path_identifier(source: str) -> str: return source -def _mapping_key_requires_redaction(key: Any) -> bool: - if isinstance(key, str): - if _mapping_key_is_direct_sensitive_assignment(key): - return True - if redact_source_identifier(key) != key: - return False - if _is_sensitive_export_key(key): - return True - if isinstance(key, bytes): - try: - key.decode("utf-8") - except UnicodeDecodeError: - return True - return False - return not (isinstance(key, (str, int, float, bool)) or key is None) - - -def _mapping_key_is_direct_sensitive_assignment(key: str) -> bool: - normalized_key = _normalize_escaped_url_delimiters_for_display(key) - decoded_key, _ = _bounded_unquote(normalized_key) - stripped_key = decoded_key.lstrip() - for pattern in (_EXPORT_EQUALS_KEY_RE, _EXPORT_HEADER_KEY_RE, _EXPORT_ENCODED_SEPARATOR_RE): - match = pattern.match(stripped_key) - if match is not None and _is_sensitive_export_key(match.group("key")): - return True - return False - - def _has_local_userinfo_credentials(path: str, *, allow_username_only: bool = False) -> bool: for segment in re.split(r"[\\/]", path): decoded_segment, decode_incomplete = _bounded_unquote(segment) @@ -1071,30 +918,3 @@ def _filter_url_params(value: str, *, preserve_redacted_params: bool) -> str: continue safe_parts.append(part) return "&".join(safe_parts) - - -def _redact_mapping_key(value: Any) -> str | int | float | bool | None: - redacted = redact_source_value(value) - if isinstance(redacted, (str, int, float, bool)) or redacted is None: - return redacted - return redact_source_text(str(redacted)) - - -def _unique_redacted_mapping_key( - key: Any, - mapping: dict[Any, Any], - *, - next_occurrences: dict[str, int], -) -> Any: - """Preserve entries whose credential-safe mapping keys collide.""" - if key not in mapping: - next_occurrences.setdefault(str(key), 2) - return key - base_key = str(key) - occurrence = next_occurrences.get(base_key, 2) - candidate = f"{base_key}#modelaudit-redacted-key-{occurrence}" - while candidate in mapping: - occurrence += 1 - candidate = f"{base_key}#modelaudit-redacted-key-{occurrence}" - next_occurrences[base_key] = occurrence + 1 - return candidate diff --git a/modelaudit/integrations/_sarif_identity_urls.py b/modelaudit/integrations/_sarif_identity_urls.py new file mode 100644 index 000000000..9dd78c272 --- /dev/null +++ b/modelaudit/integrations/_sarif_identity_urls.py @@ -0,0 +1,362 @@ +"""URL normalization retained for historical SARIF identity compatibility.""" + +import re +from urllib.parse import unquote_plus, urlsplit, urlunsplit + +_QUERY_PARAM_RE = re.compile(r"(?P[?&#;])(?P[^=\s&#;]+)=(?P[^\s&#;]*)") + +_BARE_ASSIGNMENT_RE = re.compile( + r"""(?[0-9A-Za-z_%.-]+)(?P\s*=\s*)(?![=])""" + r"""(?P"[^"\r\n]*"|'[^'\r\n]*'|"[^"\r\n]*|'[^'\r\n]*|""" + r"""(?:(?:bearer|basic|digest|negotiate|token|aws4-hmac-sha256)\s+)?[^\s&#;,)}\]]+)""", + re.IGNORECASE, +) + +_HEADER_KEY_RE = re.compile( + r"(?[0-9A-Za-z_%.-]+)\s*:", + re.IGNORECASE, +) + +_URL_USERINFO_RE = re.compile(r"([a-z][a-z0-9+.-]*://)([^/@\s]+)@", re.IGNORECASE) + +_URL_TEXT_CHARACTER = r'(?:[^\s"\'<>]||)' + +_URL_TOKEN_RE = re.compile( + rf"(?/|u002f|u003a|u003f|u003d|u0026|u0023|u003b|x2f|x3a|x3f|x3d|x26|x23|x3b)", + re.IGNORECASE, +) + +_PERCENT_ENCODED_URL_DELIMITER_RE = re.compile( + r"%(?:25)*(?P3f|3d|26|23|3b)", + re.IGNORECASE, +) + +_PERCENT_ENCODED_URL_BOUNDARY_RE = re.compile(r"%(?:25)*(?:3f|23|3b)", re.IGNORECASE) + +_PERCENT_ENCODED_URL_PREFIX_RE = re.compile( + r"(?[a-z][a-z0-9+.-]*)(?:%(?:25)*3a|:)(?:%(?:25)*2f|/)(?:%(?:25)*2f|/)", + re.IGNORECASE, +) + +_PERCENT_ENCODED_AUTHORITY_DELIMITER_RE = re.compile( + r"%(?:25)*(?P3a|40|5b|5d)", + re.IGNORECASE, +) + +_PERCENT_ENCODED_SLASH_RE = re.compile(r"%(?:25)*2f", re.IGNORECASE) + +_SAFE_DISPLAY_QUERY_KEYS = frozenset( + { + "campaign", + "download", + "lang", + "language", + "locale", + "page", + "section", + "tokenizer", + "visible", + } +) + +_MAX_QUERY_VALUE_DECODE_PASSES = 4 + +_SENSITIVE_ASSIGNMENT_KEY_TOKENS = frozenset( + { + "auth", + "authorization", + "credential", + "credentials", + "password", + "passwd", + "sas", + "secret", + "session", + "sig", + "signature", + "token", + } +) + +_SENSITIVE_ASSIGNMENT_KEY_MARKERS = ( + "accesskey", + "accesstoken", + "apikey", + "authkey", + "authtoken", + "clientsecret", + "privatekey", + "securitytoken", +) + + +def is_stream_url(url: str) -> bool: + """Return True for a stream source identifier, regardless of scheme casing.""" + return url[:9].casefold() == "stream://" + + +def redact_url_for_display(url: str) -> str: + """Remove credentials, query strings, and fragments from a URL for display.""" + try: + normalized_url = _normalize_percent_encoded_url_delimiters_for_display( + normalize_escaped_url_delimiters_for_display(url) + ) + normalized_url = _normalize_percent_encoded_url_authority_for_display(normalized_url) + parts = urlsplit(normalized_url) + if not parts.scheme: + return url + + hostname = parts.hostname or "" + netloc = f"[{hostname}]" if ":" in hostname else hostname + if parts.port is not None: + netloc = f"{netloc}:{parts.port}" + + safe_path = _strip_url_path_assignments_for_display(parts.path) + return urlunsplit((parts.scheme, netloc, safe_path, "", "")) + except Exception: + return "" + + +def redact_cloud_error_for_display(message: object, source_url: str | None = None) -> str: + """Remove signed URL credentials from provider exception text.""" + redacted = _normalize_percent_encoded_url_delimiters_for_display( + normalize_escaped_url_delimiters_for_display(str(message)) + ) + if source_url: + normalized_source_url = normalize_escaped_url_delimiters_for_display(source_url) + redacted = redacted.replace(normalized_source_url, redact_url_for_display(normalized_source_url)) + redacted = _URL_TOKEN_RE.sub(lambda match: _redact_embedded_url_for_display(match.group(0)), redacted) + redacted = _URL_USERINFO_RE.sub(r"\1@", redacted) + redacted = _BARE_ASSIGNMENT_RE.sub(_redact_bare_sensitive_assignment, redacted) + redacted = _QUERY_PARAM_RE.sub(_redact_sensitive_query_param, redacted) + return _redact_sensitive_header_assignments(redacted) + + +def normalize_escaped_url_delimiters_for_display(value: str) -> str: + """Expose backslash-escaped URL structure so reporting redactors can inspect it.""" + replacements = { + "/": "/", + "u002f": "/", + "u003a": ":", + "u003f": "?", + "u003d": "=", + "u0026": "&", + "u0023": "#", + "u003b": ";", + "x2f": "/", + "x3a": ":", + "x3f": "?", + "x3d": "=", + "x26": "&", + "x23": "#", + "x3b": ";", + } + normalized = _ESCAPED_URL_DELIMITER_RE.sub( + lambda match: replacements[match.group("delimiter").lower()], + value, + ) + return _PERCENT_ENCODED_URL_PREFIX_RE.sub(lambda match: f"{match.group('scheme')}://", normalized) + + +def _normalize_percent_encoded_url_delimiters_for_display(url: str) -> str: + """Expose encoded query structure without decoding ordinary path escapes.""" + percent_replacements = { + "3f": "?", + "3d": "=", + "26": "&", + "23": "#", + "3b": ";", + } + + def normalize_token(match: re.Match[str]) -> str: + token = match.group(0) + for boundary in _PERCENT_ENCODED_URL_BOUNDARY_RE.finditer(token): + decoded_suffix = _PERCENT_ENCODED_URL_DELIMITER_RE.sub( + lambda delimiter_match: percent_replacements[delimiter_match.group("delimiter").lower()], + token[boundary.start() :], + ) + next_major_boundary = len(decoded_suffix) + for delimiter in "?#": + delimiter_index = decoded_suffix.find(delimiter, 1) + if delimiter_index >= 0: + next_major_boundary = min(next_major_boundary, delimiter_index) + if _QUERY_PARAM_RE.search(decoded_suffix[:next_major_boundary]): + return f"{token[: boundary.start()]}{decoded_suffix}" + return token + + return re.sub(r"""[^\s"'<>]+""", normalize_token, url) + + +def _normalize_percent_encoded_url_authority_for_display(url: str) -> str: + """Expose encoded authority separators without decoding ordinary path escapes.""" + scheme_end = url.find("://") + if scheme_end < 0: + return url + + authority_start = scheme_end + 3 + authority_end = len(url) + for delimiter in "/?#": + delimiter_index = url.find(delimiter, authority_start) + if delimiter_index >= 0: + authority_end = min(authority_end, delimiter_index) + + authority = url[authority_start:authority_end] + replacements = {"3a": ":", "40": "@", "5b": "[", "5d": "]"} + normalized_authority = _PERCENT_ENCODED_AUTHORITY_DELIMITER_RE.sub( + lambda match: replacements[match.group("delimiter").lower()], + authority, + ) + if "@" in normalized_authority: + userinfo, host_and_path = normalized_authority.rsplit("@", 1) + normalized_authority = f"{userinfo}@{_PERCENT_ENCODED_SLASH_RE.sub('/', host_and_path)}" + else: + normalized_authority = _PERCENT_ENCODED_SLASH_RE.sub("/", normalized_authority) + return f"{url[:authority_start]}{normalized_authority}{url[authority_end:]}" + + +def _redact_embedded_url_for_display(url: str) -> str: + if is_stream_url(url): + return f"stream://{redact_stream_url_for_display(url[9:])}" + url = _normalize_percent_encoded_url_delimiters_for_display(url) + url = _normalize_percent_encoded_url_authority_for_display(url) + try: + parts = urlsplit(url) + safe_base = redact_url_for_display(url) + if safe_base == "": + return safe_base + safe_parts = urlsplit(safe_base) + except Exception: + return "" + + safe_query = _redact_url_component_for_display(parts.query) + safe_fragment = _redact_url_component_for_display(parts.fragment) + return urlunsplit((safe_parts.scheme, safe_parts.netloc, safe_parts.path, safe_query, safe_fragment)) + + +def _redact_url_component_for_display(value: str) -> str: + safe_parts: list[str] = [] + for part in re.split(r"[&;]", value): + if "=" not in part: + continue + key, parameter_value = part.split("=", 1) + if _is_safe_display_query_param(key, parameter_value): + safe_parts.append(part) + else: + safe_parts.append(f"{key}=") + return "&".join(safe_parts) + + +def _redact_bare_sensitive_assignment(match: re.Match[str]) -> str: + key = match.group("key") + if not _is_sensitive_assignment_key(key): + return match.group(0) + return f"{key}{match.group('separator')}" + + +def _redact_sensitive_header_assignments(value: str) -> str: + matches = [match for match in _HEADER_KEY_RE.finditer(value) if _is_sensitive_assignment_key(match.group("key"))] + for match in reversed(matches): + value_end = len(value) + for delimiter in ("\r", "\n", ",", ";"): + delimiter_index = value.find(delimiter, match.end()) + if delimiter_index >= 0: + value_end = min(value_end, delimiter_index) + value = f"{value[: match.start()]}{match.group('key')}: {value[value_end:]}" + return value + + +def _strip_url_path_assignments_for_display(path: str) -> str: + safe_segments: list[str] = [] + for segment in path.split("/"): + base, *parameters = segment.split(";") + safe_parameters = [parameter for parameter in parameters if "=" not in parameter] + safe_segments.append(";".join((base, *safe_parameters))) + return "/".join(safe_segments) + + +def _is_sensitive_assignment_key(key: str) -> bool: + decoded_key = key + for _ in range(_MAX_QUERY_VALUE_DECODE_PASSES): + next_key = unquote_plus(decoded_key) + if next_key == decoded_key: + break + decoded_key = next_key + else: + if unquote_plus(decoded_key) != decoded_key: + return True + + normalized_key = decoded_key.casefold() + key_tokens = {token for token in re.split(r"[^a-z0-9]+", normalized_key) if token} + if key_tokens & _SENSITIVE_ASSIGNMENT_KEY_TOKENS: + return True + collapsed_key = re.sub(r"[^a-z0-9]+", "", normalized_key) + return any(marker in collapsed_key for marker in _SENSITIVE_ASSIGNMENT_KEY_MARKERS) + + +def is_sensitive_credential_key(key: object) -> bool: + """Return whether a structured metadata key identifies credential material.""" + if isinstance(key, bytes): + try: + key = key.decode("utf-8") + except UnicodeDecodeError: + return False + return isinstance(key, str) and _is_sensitive_assignment_key(key) + + +def _redact_sensitive_query_param(match: re.Match[str]) -> str: + if _is_safe_display_query_param(match.group("key"), match.group("value")): + return match.group(0) + return f"{match.group('prefix')}{match.group('key')}=" + + +def _is_safe_display_query_param(key: str, value: str) -> bool: + decoded_key = unquote_plus(key).lower() + if decoded_key not in _SAFE_DISPLAY_QUERY_KEYS: + return False + + decoded_value = value + for _ in range(_MAX_QUERY_VALUE_DECODE_PASSES): + if _has_unsafe_display_query_value_structure(decoded_value): + return False + next_value = unquote_plus(decoded_value) + if next_value == decoded_value: + return True + decoded_value = next_value + + # Values that remain encoded after the bounded pass may conceal nested + # query structure under additional encoding layers. + return not _has_unsafe_display_query_value_structure(decoded_value) and unquote_plus(decoded_value) == decoded_value + + +def _has_unsafe_display_query_value_structure(value: str) -> bool: + return any(delimiter in value for delimiter in "?&#;=") or any( + ord(character) < 0x20 or ord(character) == 0x7F for character in value + ) + + +def redact_stream_url_for_display(url: str) -> str: + """Return a fail-closed display identifier for a stream source URL.""" + try: + if not urlsplit(url).scheme: + return "" + except Exception: + return "" + return redact_url_for_display(url) + + +def redact_stream_error_for_display(message: object, source_url: str) -> str: + """Remove a stream source URL from exception text, including malformed identifiers.""" + safe_url = redact_stream_url_for_display(source_url) + redacted = str(message) + if not source_url: + return redact_cloud_error_for_display(redacted.replace("stream://", f"stream://{safe_url}")) + redacted = redacted.replace(f"stream://{source_url}", f"stream://{safe_url}") + redacted = redacted.replace(source_url, safe_url) + return redact_cloud_error_for_display(redacted) diff --git a/modelaudit/integrations/jfrog.py b/modelaudit/integrations/jfrog.py index e9d37b44e..44b4aa114 100644 --- a/modelaudit/integrations/jfrog.py +++ b/modelaudit/integrations/jfrog.py @@ -13,12 +13,12 @@ from ..models import ModelAuditResultModel from ..scanner_selection import SCANNER_SELECTION_CONFIG_KEY +from ..utils.helpers.evidence import format_terminal_text from ..utils.sources.jfrog import ( detect_jfrog_target_type, download_artifact, download_jfrog_folder, format_size, - redact_jfrog_url_for_display, ) logger = logging.getLogger(__name__) @@ -144,11 +144,12 @@ def scan_jfrog_artifact( scan_cache_dir = str(Path(raw_cache_dir).expanduser()) if cache_enabled and raw_cache_dir else None download_dir, cleanup_download_dir = _prepare_download_dir(url, scan_cache_dir) start_time = time.time() - display_url = redact_jfrog_url_for_display(url) + display_url = url + log_url = format_terminal_text(url) try: # Detect if URL points to a file or folder - logger.debug(f"Analyzing JFrog target {display_url}") + logger.debug(f"Analyzing JFrog target {log_url}") target_info = detect_jfrog_target_type( url, api_token=api_token, @@ -159,7 +160,7 @@ def scan_jfrog_artifact( if target_info["type"] == "file": file_download_limit = _positive_limit(max_download_size, max_file_size, max_total_size) _require_known_file_size_within_limit(target_info, limit=file_download_limit, display_url=display_url) - logger.debug(f"Downloading JFrog file {display_url} to {download_dir}") + logger.debug(f"Downloading JFrog file {log_url} to {download_dir}") download_path = download_artifact( url, cache_dir=download_dir, @@ -169,7 +170,7 @@ def scan_jfrog_artifact( max_size=file_download_limit, ) else: - logger.debug(f"Downloading JFrog folder {display_url} to {download_dir}") + logger.debug(f"Downloading JFrog folder {log_url} to {download_dir}") folder_download_kwargs: dict[str, Any] = {} if scannable_extensions is not None: folder_download_kwargs["scannable_extensions"] = scannable_extensions diff --git a/modelaudit/integrations/mlflow.py b/modelaudit/integrations/mlflow.py index 73d59060e..84f1c3d02 100644 --- a/modelaudit/integrations/mlflow.py +++ b/modelaudit/integrations/mlflow.py @@ -14,16 +14,18 @@ from typing import Any from urllib.parse import unquote, urlparse -from ..detectors.network_comm import _redact_urls_in_text -from ..models import Check, CheckStatus, Issue, IssueSeverity, ModelAuditResultModel, create_initial_audit_result -from ..scanners._evidence_redaction import ( - MAX_PERCENT_DECODE_PASSES, - MAX_REDACTION_VALUE_DEPTH, - SENSITIVE_CONTAINER_KEY, - redact_evidence_string, - redact_evidence_value, +from ..models import ( + Check, + CheckStatus, + FileMetadataModel, + Issue, + IssueSeverity, + ModelAuditResultModel, + create_initial_audit_result, ) -from ..utils.sources.cloud_storage import redact_cloud_error_for_display, redact_url_for_display +from ..utils.helpers.evidence import format_evidence_string, format_evidence_value, format_terminal_text +from ..utils.helpers.finding_identity import preserve_finding_identity +from ._mlflow_identity import mlflow_source_identity logger = logging.getLogger(__name__) @@ -52,33 +54,6 @@ _MLFLOW_PERCENT_ESCAPE_RE = re.compile(r"%[0-9A-Fa-f]{2}") _OS_OPEN_SUPPORTS_DIR_FD = os.open in os.supports_dir_fd _IS_WINDOWS = os.name == "nt" -_MLFLOW_SENSITIVE_KEY = rf"(?:{SENSITIVE_CONTAINER_KEY}|credentials?|jwt|session)" -_MLFLOW_SENSITIVE_ASSIGNMENT_RE = re.compile( - r"(?ix)" - r"(?P(?\"(?:\\.|[^\"\\])*\"|'(?:\\.|[^'\\])*'|(?:(?:bearer|basic|token)\s+)?[^\s,;&}\]]+)" -) -_MLFLOW_BRACKETED_SENSITIVE_ASSIGNMENT_RE = re.compile( - r"(?ix)" - r"(?P(?\"(?:\\.|[^\"\\])*\"|'(?:\\.|[^'\\])*'|(?:(?:bearer|basic|token)\s+)?[^\s,;&}\]]+)" -) -_MLFLOW_SENSITIVE_CONTAINER_PREFIX_RE = re.compile( - r"(?ix)" - r"(?P(?[({\[])", -) -_MLFLOW_PROTOCOL_RELATIVE_URL_RE = re.compile( - r"(?i)(?:(?:[\\/]|%(?:25)*(?:2f|5c)){2,})[^\s\"'<>]+", -) -_MLFLOW_BENIGN_AUTH_CONTEXT_RE = re.compile( - r"(?i)\b(?:bearer|basic|token)(?=\s+(?:authentication|endpoint|refresh|service)\b)", -) @dataclass(frozen=True) @@ -197,14 +172,25 @@ def _split_mlflow_artifact_uri(mlflow_module: Any, model_uri: str) -> tuple[str, return f"models:/{parts[0]}/{parts[1]}", "/".join(parts[2:]) +def _mlflow_report_source_identifier(model_uri: str) -> str: + """Keep lossy source previews distinct without exceeding the evidence bound.""" + marker = "#modelaudit-source-sha256-" + preview = format_evidence_string(model_uri, max_chars=_MAX_MLFLOW_ERROR_DISPLAY_CHARS) + if preview == model_uri and marker not in model_uri: + return preview + # Reserve the suffix syntax so a literal URI cannot impersonate a generated identifier. + digest = hashlib.sha256(model_uri.encode("utf-8", errors="surrogatepass")).hexdigest() + return preview[: _MAX_MLFLOW_ERROR_DISPLAY_CHARS - len(marker) - len(digest)] + marker + digest + + def _mlflow_budget_failure_result(model_uri: str, message: str, details: dict[str, Any]) -> ModelAuditResultModel: result = create_initial_audit_result() result.scanner_names = ["mlflow"] result.has_errors = True result.success = False - safe_model_uri = _redact_mlflow_detail_value_for_display(model_uri) - safe_details = redact_evidence_value( - _redact_mlflow_detail_value_for_display(details), + safe_model_uri = _mlflow_report_source_identifier(model_uri) + safe_details = format_evidence_value( + details, max_string_chars=_MAX_MLFLOW_ERROR_DISPLAY_CHARS, ) @@ -233,6 +219,12 @@ def _mlflow_budget_failure_result(model_uri: str, message: str, details: dict[st type="mlflow_download_budget", ) ) + identity_location = mlflow_source_identity(model_uri) + for record in [result.issues[-1], result.checks[-1]]: + preserve_finding_identity(record, "mlflow_acquisition", location=identity_location) + result.file_metadata[safe_model_uri] = FileMetadataModel( + source_identity={"producer": "mlflow_acquisition", "path": identity_location} + ) result.finalize_statistics() return result @@ -246,9 +238,9 @@ def _mlflow_artifact_trust_failure_result( result.scanner_names = ["mlflow"] result.has_errors = True result.success = False - safe_model_uri = _redact_mlflow_detail_value_for_display(model_uri) - safe_details = redact_evidence_value( - _redact_mlflow_detail_value_for_display(details), + safe_model_uri = _mlflow_report_source_identifier(model_uri) + safe_details = format_evidence_value( + details, max_string_chars=_MAX_MLFLOW_ERROR_DISPLAY_CHARS, ) @@ -278,6 +270,12 @@ def _mlflow_artifact_trust_failure_result( type=_MLFLOW_ARTIFACT_TRUST_FAILURE_TYPE, ) ) + identity_location = mlflow_source_identity(model_uri) + for record in [result.issues[-1], result.checks[-1]]: + preserve_finding_identity(record, "mlflow_acquisition", location=identity_location) + result.file_metadata[safe_model_uri] = FileMetadataModel( + source_identity={"producer": "mlflow_acquisition", "path": identity_location} + ) result.finalize_statistics() return result @@ -291,9 +289,9 @@ def _mlflow_download_safety_failure_result( result.scanner_names = ["mlflow"] result.has_errors = True result.success = False - safe_model_uri = _redact_mlflow_detail_value_for_display(model_uri) - safe_details = redact_evidence_value( - _redact_mlflow_detail_value_for_display(details), + safe_model_uri = _mlflow_report_source_identifier(model_uri) + safe_details = format_evidence_value( + details, max_string_chars=_MAX_MLFLOW_ERROR_DISPLAY_CHARS, ) @@ -325,6 +323,12 @@ def _mlflow_download_safety_failure_result( type="mlflow_download_path", ) ) + identity_location = mlflow_source_identity(model_uri) + for record in [result.issues[-1], result.checks[-1]]: + preserve_finding_identity(record, "mlflow_acquisition", location=identity_location) + result.file_metadata[safe_model_uri] = FileMetadataModel( + source_identity={"producer": "mlflow_acquisition", "path": identity_location} + ) result.finalize_statistics() return result @@ -705,147 +709,8 @@ def _trusted_mlflow_delegated_download_plan( return _MlflowDelegatedDownloadPlan(remote_targets, local_plan) -def _redact_mlflow_error_for_display(error: object) -> str: - def _replace_sensitive_value(match: re.Match[str]) -> str: - value = match.group("value") - quote = value[0] if value[:1] in {'"', "'"} else "" - return f"{match.group('prefix')}{quote}{quote}" - - def _redact_protocol_relative_url(match: re.Match[str]) -> str: - candidate = match.group(0) - decoded = candidate - for _ in range(MAX_PERCENT_DECODE_PASSES): - next_decoded = unquote(decoded) - if next_decoded == decoded: - break - decoded = next_decoded - - normalized = decoded.replace("\\/", "/").replace("\\", "/") - if len(normalized) - len(normalized.lstrip("/")) < 2: - return candidate - normalized = f"//{normalized.lstrip('/')}" - authority = normalized[2:].split("/", 1)[0].split("?", 1)[0].split("#", 1)[0] - if "@" not in authority: - return candidate - - safe_url = redact_url_for_display(f"https:{normalized}") - return safe_url.removeprefix("https:") - - def _redact_sensitive_containers(text: str) -> str: - parts: list[str] = [] - cursor = 0 - closing_delimiters = {"(": ")", "[": "]", "{": "}"} - - while match := _MLFLOW_SENSITIVE_CONTAINER_PREFIX_RE.search(text, cursor): - parts.append(text[cursor : match.start()]) - parts.append(f"{match.group('prefix')}") - stack = [closing_delimiters[match.group("open")]] - quote: str | None = None - escaped = False - index = match.end() - - while index < len(text) and stack: - character = text[index] - if quote is not None: - if escaped: - escaped = False - elif character == "\\": - escaped = True - elif character == quote: - quote = None - elif character in {'"', "'"}: - quote = character - elif character in closing_delimiters: - if len(stack) >= MAX_REDACTION_VALUE_DEPTH: - index = len(text) - break - stack.append(closing_delimiters[character]) - elif character == stack[-1]: - stack.pop() - index += 1 - - if stack: - cursor = len(text) - break - cursor = index - - parts.append(text[cursor:]) - return "".join(parts) - - redacted = _MLFLOW_PROTOCOL_RELATIVE_URL_RE.sub(_redact_protocol_relative_url, str(error)) - redacted = _redact_sensitive_containers(redacted) - redacted = _MLFLOW_BRACKETED_SENSITIVE_ASSIGNMENT_RE.sub(_replace_sensitive_value, redacted) - redacted = _MLFLOW_SENSITIVE_ASSIGNMENT_RE.sub(_replace_sensitive_value, redacted) - contains_url = bool( - re.search(r"(?i)(?:\b[a-z][a-z0-9+.-]*://|\bmodels:/)", redacted) - or _MLFLOW_PROTOCOL_RELATIVE_URL_RE.search(redacted) - ) - if contains_url: - redacted = redact_cloud_error_for_display(_redact_urls_in_text(redacted)) - redacted = _MLFLOW_PROTOCOL_RELATIVE_URL_RE.sub(_redact_protocol_relative_url, redacted) - redacted = _MLFLOW_BRACKETED_SENSITIVE_ASSIGNMENT_RE.sub(_replace_sensitive_value, redacted) - redacted = _MLFLOW_SENSITIVE_ASSIGNMENT_RE.sub(_replace_sensitive_value, redacted) - - benign_auth_contexts: list[tuple[str, str]] = [] - - def _protect_benign_auth_context(match: re.Match[str]) -> str: - placeholder = f"MODELAUDITMLFLOWSAFECONTEXT{len(benign_auth_contexts)}" - benign_auth_contexts.append((placeholder, match.group(0))) - return placeholder - - redacted = _MLFLOW_BENIGN_AUTH_CONTEXT_RE.sub(_protect_benign_auth_context, redacted) - if contains_url: - redacted = redact_evidence_string(redacted, max_chars=None) - else: - redacted = "&".join(redact_evidence_string(part, max_chars=None) for part in redacted.split("&")) - for placeholder, original in benign_auth_contexts: - redacted = redacted.replace(placeholder, original) - - if len(redacted) <= _MAX_MLFLOW_ERROR_DISPLAY_CHARS: - return redacted - return f"{redacted[: _MAX_MLFLOW_ERROR_DISPLAY_CHARS - 3]}..." - - -def _mlflow_text_requires_specialized_redaction(text: str) -> bool: - if "models:/" in text.lower(): - return True - if _MLFLOW_BRACKETED_SENSITIVE_ASSIGNMENT_RE.search(text) or _MLFLOW_SENSITIVE_CONTAINER_PREFIX_RE.search(text): - return True - for match in _MLFLOW_PROTOCOL_RELATIVE_URL_RE.finditer(text): - candidate = match.group(0) - if candidate.startswith("//") and re.search(r"(?i)[a-z][a-z0-9+.-]*:$", text[: match.start()]): - continue - decoded = candidate - for _ in range(MAX_PERCENT_DECODE_PASSES): - next_decoded = unquote(decoded) - if next_decoded == decoded: - break - decoded = next_decoded - normalized = decoded.replace("\\/", "/").replace("\\", "/") - authority = normalized.lstrip("/").split("/", 1)[0].split("?", 1)[0].split("#", 1)[0] - if "@" in authority: - return True - return False - - -def _redact_mlflow_detail_value_for_display(value: Any, *, depth: int = 0) -> Any: - if depth >= MAX_REDACTION_VALUE_DEPTH: - return "" - if isinstance(value, str): - specialized = ( - _redact_mlflow_error_for_display(value) if _mlflow_text_requires_specialized_redaction(value) else value - ) - return redact_evidence_string(specialized, max_chars=_MAX_MLFLOW_ERROR_DISPLAY_CHARS) - if isinstance(value, dict): - return { - key: _redact_mlflow_detail_value_for_display(nested_value, depth=depth + 1) - for key, nested_value in value.items() - } - if isinstance(value, list): - return [_redact_mlflow_detail_value_for_display(item, depth=depth + 1) for item in value] - if isinstance(value, tuple): - return tuple(_redact_mlflow_detail_value_for_display(item, depth=depth + 1) for item in value) - return value +def _format_mlflow_error(error: object) -> str: + return format_evidence_string(str(error), max_chars=_MAX_MLFLOW_ERROR_DISPLAY_CHARS) def _terminal_mlflow_artifact_repository(artifact_repository: Any) -> Any | None: @@ -1385,7 +1250,7 @@ def _preflight_local_mlflow_sources( details.update( { "reason": "artifact_size_unavailable", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), } ) return _mlflow_budget_failure_result( @@ -1461,7 +1326,7 @@ def _preflight_mlflow_download_budget( raise ValueError("Artifact repository wrapper chain could not be resolved") except Exception as e: details["reason"] = "artifact_size_unavailable" - details["error"] = _redact_mlflow_error_for_display(e) + details["error"] = _format_mlflow_error(e) return _mlflow_budget_failure_result( model_uri, "Unable to determine MLflow artifact size before download", @@ -1475,7 +1340,7 @@ def _preflight_mlflow_download_budget( details.update( { "reason": "artifact_size_unavailable", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), } ) return _mlflow_budget_failure_result( @@ -1671,7 +1536,7 @@ def _preflight_mlflow_download_budget( ) except Exception as e: details["reason"] = "artifact_size_unavailable" - details["error"] = _redact_mlflow_error_for_display(e) + details["error"] = _format_mlflow_error(e) return _mlflow_budget_failure_result( model_uri, "Unable to determine MLflow artifact size before download", @@ -1702,7 +1567,7 @@ def _preflight_mlflow_download_budget( { "reason": "artifact_size_unavailable", "artifact_file_count": 0, - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), } ) return _mlflow_budget_failure_result( @@ -1797,7 +1662,7 @@ def _download_preflighted_mlflow_artifacts( details.update( { "reason": "artifact_download_verification_failed", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), } ) return _mlflow_budget_failure_result( @@ -1846,7 +1711,7 @@ def _download_preflighted_mlflow_artifacts( { "reason": "artifact_download_verification_failed", "artifact_path": artifact.path, - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), } ) return _mlflow_budget_failure_result( @@ -1909,7 +1774,7 @@ def _download_preflighted_mlflow_artifacts( details.update( { "reason": "artifact_download_verification_failed", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), } ) return _mlflow_budget_failure_result( @@ -2083,7 +1948,7 @@ def _download_trusted_mlflow_artifacts( "MLflow artifact repository returned an unsafe artifact listing", { "reason": "artifact_listing_unsafe", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), }, ) if target.optional_when_missing and not artifact_paths: @@ -2255,7 +2120,7 @@ def _capture_mlflow_download_root( "Unable to establish the MLflow staging directory", { "reason": "artifact_download_root_unavailable", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), }, ) @@ -2288,7 +2153,7 @@ def _capture_mlflow_download_root( "Unable to hold the MLflow staging directory open", { "reason": "artifact_download_root_unavailable", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), }, ) @@ -2415,7 +2280,7 @@ def _validate_mlflow_download_tree( "Unable to validate the MLflow staging directory", { "reason": "artifact_download_path_unavailable", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), }, ) @@ -2447,7 +2312,7 @@ def _resolve_mlflow_download_path( "Unable to verify MLflow artifact download path", { "reason": "artifact_download_path_unavailable", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), }, ) @@ -2463,7 +2328,7 @@ def _resolve_mlflow_download_path( "Unable to verify the MLflow staging directory identity", { "reason": "artifact_download_root_unavailable", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), }, ) @@ -2505,7 +2370,7 @@ def _resolve_mlflow_download_path( "Unable to inspect MLflow artifact download path", { "reason": "artifact_download_path_unavailable", - "error": _redact_mlflow_error_for_display(exc), + "error": _format_mlflow_error(exc), }, ) @@ -2609,7 +2474,9 @@ def scan_mlflow_model( return captured_download_root download_root_identity = captured_download_root - logger.debug(f"Downloading MLflow model {_redact_mlflow_error_for_display(model_uri)} to {download_dir}") + logger.debug( + f"Downloading MLflow model {format_terminal_text(_format_mlflow_error(model_uri))} to {download_dir}" + ) local_path: str | None if isinstance(download_plan, _MlflowDownloadPlan): download_result = _download_preflighted_mlflow_artifacts( diff --git a/modelaudit/integrations/sarif_formatter.py b/modelaudit/integrations/sarif_formatter.py index 264faabc3..56f84f1ad 100644 --- a/modelaudit/integrations/sarif_formatter.py +++ b/modelaudit/integrations/sarif_formatter.py @@ -6,31 +6,36 @@ import contextlib import json +from collections.abc import Callable, Iterable from pathlib import Path from typing import Any from urllib.parse import quote +from pydantic import TypeAdapter + from modelaudit import __version__ from modelaudit.core_results import ( determine_exit_code, results_have_inconclusive_outcome, results_have_operational_error, ) -from modelaudit.integrations.source_redaction import ( - redact_prevalidated_source_value as _redact_prevalidated_value_for_sarif, -) -from modelaudit.integrations.source_redaction import ( - redact_source_identifier as _redact_path_for_sarif, +from modelaudit.integrations._sarif_identity import ( + redact_source_identifier as _identity_location, ) -from modelaudit.integrations.source_redaction import ( - redact_source_text as _redact_text_for_sarif, +from modelaudit.integrations._sarif_identity import ( + redact_source_text as _identity_text, ) -from modelaudit.integrations.source_redaction import ( - redact_source_value as _redact_value_for_sarif, +from modelaudit.integrations.source_serialization import ( + serialize_source_identifier, + serialize_source_text, + serialize_source_value, + serialize_source_values, ) from modelaudit.models import ModelAuditResultModel from modelaudit.scanner_results import IssueSeverity -from modelaudit.scanners._catboost_evidence_redaction import redact_evidence_string as _redact_catboost_evidence +from modelaudit.utils.helpers.finding_identity import finding_identity + +_JSON_VALUE_ADAPTER: TypeAdapter[Any] = TypeAdapter(Any) def format_sarif_output( @@ -54,7 +59,7 @@ def format_sarif_output( "runs": [_create_run(audit_result, scan_paths, verbose)], } - return json.dumps(sarif_output, indent=2) + return json.dumps(_JSON_VALUE_ADAPTER.dump_python(sarif_output, mode="json"), indent=2) def _create_run( @@ -63,23 +68,35 @@ def _create_run( verbose: bool, ) -> dict[str, Any]: """Create a SARIF run object from ModelAudit results.""" - safe_scan_paths = [_redact_path_for_sarif(path) for path in scan_paths] - # Filter issues based on verbosity issues = audit_result.issues if not verbose: issues = [i for i in issues if i.severity != IssueSeverity.DEBUG] issues = _primary_sarif_issues(issues) - + source_identifiers, serialized_properties = _serialize_sarif_sources( + [ + *scan_paths, + *(asset.path for asset in audit_result.assets), + *(issue.location for issue in issues if issue.location), + ], + issues, + ) + safe_scan_paths = [source_identifiers[path] for path in scan_paths] # Create rules from unique issue types rules = _create_rules(issues, prefiltered=True) rule_indices = {rule["id"]: idx for idx, rule in enumerate(rules)} # Create results from issues - results = _create_results(issues, rule_indices, prefiltered=True) + results = _create_results( + issues, + rule_indices, + prefiltered=True, + source_identifiers=source_identifiers, + serialized_properties=serialized_properties, + ) # Create artifacts from scanned files - artifacts = _create_artifacts(audit_result) + artifacts = _create_artifacts(audit_result, source_identifiers=source_identifiers) exit_code = determine_exit_code(audit_result) has_operational_errors = results_have_operational_error(audit_result) has_incomplete_coverage = results_have_inconclusive_outcome(audit_result) @@ -186,15 +203,16 @@ def _create_rules(issues: list, *, prefiltered: bool = False) -> list[dict[str, seen_rules = set() for issue in issues: + identity_issue = finding_identity(issue) # Create a rule ID from the issue type or message - rule_id = _get_rule_id(issue) + rule_id = _get_rule_id(identity_issue) if rule_id not in seen_rules: seen_rules.add(rule_id) rule: dict[str, Any] = { "id": rule_id, - "name": _get_rule_name(issue), + "name": _get_rule_name(identity_issue), "shortDescription": {"text": _get_rule_short_description(issue)}, "fullDescription": {"text": _get_rule_full_description(issue)}, "defaultConfiguration": { @@ -209,13 +227,13 @@ def _create_rules(issues: list, *, prefiltered: bool = False) -> list[dict[str, }, } - rule_code = _get_issue_rule_code(issue) + rule_code = _get_issue_rule_code(identity_issue) if rule_code: rule["properties"]["rule_code"] = rule_code # Add help information if available if hasattr(issue, "why") and issue.why: - redacted_why = _redact_text_for_sarif(issue.why) + redacted_why = serialize_source_text(issue.why) rule["help"] = {"text": redacted_why, "markdown": redacted_why} rules.append(rule) @@ -223,26 +241,53 @@ def _create_rules(issues: list, *, prefiltered: bool = False) -> list[dict[str, return rules +def _sarif_source_identifier_key(paths: Iterable[str]) -> Callable[[str], str] | None: + """Reserve URI equivalences only when a source needs an artificial identifier.""" + return ( + _source_identifier_comparison_key if any(serialize_source_identifier(path) != path for path in paths) else None + ) + + +def _source_identifier_comparison_key(path: str) -> str: + """Reserve URI-equivalent names independently of the working directory.""" + return quote(Path(path).name, safe="/", errors="backslashreplace") + + +def _serialize_sarif_sources(paths: list[str], issues: list) -> tuple[dict[str, str], list[dict[str, Any]]]: + """Allocate source references together with the evidence that refers to them.""" + serialized = serialize_source_values( + [paths, *(dict(issue.details or {}) for issue in issues)], identifier_key=_sarif_source_identifier_key(paths) + ) + return dict(zip(paths, serialized[0], strict=True)), serialized[1:] + + def _create_results( issues: list, rule_indices: dict[str, int] | None = None, *, prefiltered: bool = False, + source_identifiers: dict[str, str] | None = None, + serialized_properties: list[dict[str, Any]] | None = None, ) -> list[dict[str, Any]]: """Create SARIF results from issues.""" if not prefiltered: issues = _primary_sarif_issues(issues) + if source_identifiers is None or serialized_properties is None: + source_identifiers, serialized_properties = _serialize_sarif_sources( + [issue.location for issue in issues if issue.location], issues + ) results = [] if rule_indices is None: rule_indices = {rule["id"]: idx for idx, rule in enumerate(_create_rules(issues, prefiltered=prefiltered))} - for issue in issues: - rule_id = _get_rule_id(issue) + for issue, properties in zip(issues, serialized_properties, strict=True): + identity_issue = finding_identity(issue) + rule_id = _get_rule_id(identity_issue) result = { "ruleId": rule_id, "ruleIndex": rule_indices[rule_id], "level": _severity_to_sarif_level(issue.severity), - "message": {"text": _redact_text_for_sarif(issue.message)}, + "message": {"text": serialize_source_text(issue.message)}, "locations": [], "partialFingerprints": {}, "relatedLocations": [], @@ -256,7 +301,10 @@ def _create_results( if issue.location: location = { "physicalLocation": { - "artifactLocation": {"uri": _normalize_path_to_uri(issue.location), "uriBaseId": "%SRCROOT%"} + "artifactLocation": { + "uri": _normalize_path_to_uri(issue.location, source_identifiers), + "uriBaseId": "%SRCROOT%", + } } } @@ -273,75 +321,53 @@ def _create_results( import hashlib fingerprint = "" - fingerprint_location = _redact_path_for_sarif(issue.location or "") + fingerprint_message = identity_issue.message + fingerprint_location = identity_issue.location or "" + fingerprint_location = _identity_location(fingerprint_location) if issue.details: - evidence_fingerprint = _redact_text_for_sarif(str(issue.details.get("evidence_fingerprint", ""))) + evidence_fingerprint = _identity_text(str(identity_issue.details.get("evidence_fingerprint", ""))) if evidence_fingerprint: fingerprint = hashlib.sha256( "\x1f".join((evidence_fingerprint, fingerprint_location, str(issue.severity))).encode() ).hexdigest()[:16] if not fingerprint: - fingerprint_message = _redact_text_for_sarif(issue.message) + fingerprint_message = _identity_text(fingerprint_message) fingerprint = hashlib.sha256( f"{fingerprint_message}{fingerprint_location}{issue.severity}".encode() ).hexdigest()[:16] result["partialFingerprints"]["primaryLocationLineHash"] = fingerprint # type: ignore[index] # Add properties with additional details - # Revalidate CatBoost evidence before trusting its redaction markers, then - # apply source URL protection without discarding sanitized command context. - catboost_issue = getattr(issue, "type", None) == "catboost_check" - details = dict(issue.details or {}) - if catboost_issue: - details = _redact_catboost_details_for_sarif(details) - properties = ( - _redact_prevalidated_value_for_sarif(details) if catboost_issue else _redact_value_for_sarif(details) - ) properties.pop("rule_code", None) properties.pop("issue_type", None) - rule_code = _get_issue_rule_code(issue) + rule_code = _get_issue_rule_code(identity_issue) if rule_code: properties["rule_code"] = rule_code if hasattr(issue, "type") and issue.type: - properties["issue_type"] = _redact_text_for_sarif(issue.type) + properties["issue_type"] = serialize_source_text(issue.type) if properties: result["properties"] = properties # Add fix suggestions if available if hasattr(issue, "recommendation") and issue.recommendation: - result["fixes"] = [{"description": {"text": _redact_text_for_sarif(issue.recommendation)}}] + result["fixes"] = [{"description": {"text": serialize_source_text(issue.recommendation)}}] results.append(result) return results -def _redact_catboost_details_for_sarif(value: Any) -> Any: - """Revalidate scanner-redacted CatBoost evidence before export.""" - if isinstance(value, str): - return _redact_catboost_evidence(value, max_chars=len(value)) - if isinstance(value, dict): - return {key: _redact_catboost_details_for_sarif(item) for key, item in value.items()} - if isinstance(value, list): - return [_redact_catboost_details_for_sarif(item) for item in value] - if isinstance(value, tuple): - return tuple(_redact_catboost_details_for_sarif(item) for item in value) - if isinstance(value, set): - return {_redact_catboost_details_for_sarif(item) for item in value} - if isinstance(value, frozenset): - return frozenset(_redact_catboost_details_for_sarif(item) for item in value) - # Unknown structured values and binary evidence still take the conservative - # generic path before the prevalidated source pass sees them. - return _redact_value_for_sarif(value) - - -def _create_artifacts(audit_result: ModelAuditResultModel) -> list[dict[str, Any]]: +def _create_artifacts( + audit_result: ModelAuditResultModel, *, source_identifiers: dict[str, str] | None = None +) -> list[dict[str, Any]]: """Create SARIF artifacts from scanned files.""" artifacts: list[dict[str, Any]] = [] + if source_identifiers is None: + source_identifiers, _ = _serialize_sarif_sources([asset.path for asset in audit_result.assets], []) for asset in audit_result.assets: artifact: dict[str, Any] = { - "location": {"uri": _normalize_path_to_uri(asset.path), "uriBaseId": "%SRCROOT%"}, + "location": {"uri": _normalize_path_to_uri(asset.path, source_identifiers), "uriBaseId": "%SRCROOT%"}, "mimeType": _get_mime_type(asset.type), "properties": {"type": asset.type}, } @@ -366,7 +392,7 @@ def _create_artifacts(audit_result: ModelAuditResultModel) -> list[dict[str, Any artifact["hashes"] = hashes member_file_hashes = getattr(metadata, "member_file_hashes", None) if member_file_hashes: - artifact["properties"]["memberFileHashes"] = _redact_value_for_sarif( + artifact["properties"]["memberFileHashes"] = serialize_source_value( { member_path: ( record.model_dump(mode="json", exclude_none=True) @@ -407,11 +433,11 @@ def _get_rule_id(issue: Any) -> str: return rule_code if hasattr(issue, "type") and issue.type: - redacted_type = _redact_text_for_sarif(str(issue.type)) + redacted_type = _identity_text(str(issue.type)) return f"MA{redacted_type.replace(' ', '-').upper()}" # Generate from message if no type - redacted_message = _redact_text_for_sarif(issue.message) + redacted_message = _identity_text(issue.message) base = redacted_message[:30].replace(" ", "-").replace(":", "").upper() # Remove special characters base = "".join(c if c.isalnum() or c == "-" else "" for c in base) @@ -422,23 +448,23 @@ def _get_issue_rule_code(issue: Any) -> str | None: """Return the stable ModelAudit rule code for an issue when available.""" rule_code = getattr(issue, "rule_code", None) if isinstance(rule_code, str) and rule_code: - return _redact_text_for_sarif(rule_code) + return _identity_text(rule_code) return None def _get_rule_name(issue: Any) -> str: """Get a human-readable rule name from an issue.""" if hasattr(issue, "type") and issue.type: - return _redact_text_for_sarif(str(issue.type)).replace("_", " ").title() + return _identity_text(str(issue.type)).replace("_", " ").title() # Extract from message - redacted_message = _redact_text_for_sarif(issue.message) + redacted_message = _identity_text(issue.message) return str(redacted_message.split(":")[0] if ":" in redacted_message else redacted_message[:50]) def _get_rule_short_description(issue: Any) -> str: """Get a short description for a rule.""" - lowered_message = issue.message.lower() + lowered_message = finding_identity(issue).message.lower() if "pickle" in lowered_message: return "Potentially unsafe pickle operation detected" elif "import" in lowered_message: @@ -454,7 +480,7 @@ def _get_rule_short_description(issue: Any) -> str: elif "blacklist" in lowered_message: return "Blacklisted model name detected" else: - return str(_redact_text_for_sarif(issue.message)[:100]) + return str(serialize_source_text(issue.message)[:100]) def _get_rule_full_description(issue: Any) -> str: @@ -462,7 +488,7 @@ def _get_rule_full_description(issue: Any) -> str: desc = _get_rule_short_description(issue) if hasattr(issue, "why") and issue.why: - desc += f" {_redact_text_for_sarif(issue.why)}" + desc += f" {serialize_source_text(issue.why)}" return desc @@ -493,7 +519,7 @@ def _get_tags_for_issue(issue: Any) -> list[str]: """Get relevant tags for an issue.""" tags = ["security", "ml-model"] - message_lower = issue.message.lower() + message_lower = finding_identity(issue).message.lower() if "pickle" in message_lower: tags.append("pickle") @@ -512,9 +538,9 @@ def _get_tags_for_issue(issue: Any) -> list[str]: return tags -def _normalize_path_to_uri(path: str) -> str: +def _normalize_path_to_uri(path: str, source_identifiers: dict[str, str] | None = None) -> str: """Normalize a file path to a URI format.""" - path = _redact_path_for_sarif(path) + path = source_identifiers[path] if source_identifiers is not None else serialize_source_identifier(path) # Convert to Path object for normalization p = Path(path) @@ -526,7 +552,7 @@ def _normalize_path_to_uri(path: str) -> str: uri_path = p.as_posix() # URL-encode special characters - return quote(uri_path, safe="/") + return quote(uri_path, safe="/", errors="backslashreplace") def _get_mime_type(file_type: str) -> str: diff --git a/modelaudit/integrations/sbom_generator.py b/modelaudit/integrations/sbom_generator.py index 333225e1e..126e8593b 100644 --- a/modelaudit/integrations/sbom_generator.py +++ b/modelaudit/integrations/sbom_generator.py @@ -15,7 +15,8 @@ from ..models import FileMetadataModel, ModelAuditResultModel from ..scanner_results import Issue, IssueSeverity -from .source_redaction import redact_source_identifier, redact_source_reference, redact_source_value +from ..utils.helpers.finding_identity import finding_identity +from .source_serialization import serialize_source_identifier, serialize_source_value SCANNER_VERSION = f"v{_pkg_version('modelaudit')}" _MAX_SYMLINK_HOPS = 40 @@ -36,7 +37,7 @@ def _sbom_property_value(value: object) -> str: def _serialize_member_file_hashes(value: object) -> str: - return json.dumps(redact_source_value(value), sort_keys=True, separators=(",", ":")) + return json.dumps(serialize_source_value(value), sort_keys=True, separators=(",", ":")) def _append_member_file_hash_summary_properties(props: list[Property], metadata: FileMetadataModel) -> None: @@ -135,6 +136,74 @@ def _get_component_type(path: str, metadata: dict[str, Any] | None) -> Component return ComponentType.FILE +def _source_identity_path(path: str, metadata: FileMetadataModel | dict[str, Any] | None) -> str: + """Keep the producer's component semantics while exporting its raw source.""" + identity = metadata.get("source_identity") if isinstance(metadata, (dict, FileMetadataModel)) else None + if isinstance(identity, dict) and identity.get("producer") in ( + "stream", + "huggingface_acquisition", + "mlflow_acquisition", + ): + identity_path = identity.get("path") + if isinstance(identity_path, str): + return identity_path + return path + + +def _cli_source_classification_path(path: str) -> str: + """Preserve the CLI's historical type input without changing source evidence.""" + from urllib.parse import urlparse, urlunparse + + from ..utils.sources.cloud_storage import is_cleartext_cloud_url, is_cloud_url, is_stream_url + from ..utils.sources.huggingface_paths import _huggingface_classification_url + from ..utils.sources.jfrog import ( + _get_configured_jfrog_hosts, + _is_jfrog_service_host, + _is_local_jfrog_host, + _normalize_hostname, + ) + from ..utils.sources.pytorch_hub import is_cleartext_pytorch_hub_url, is_pytorch_hub_url + from ._sarif_identity_urls import redact_stream_url_for_display, redact_url_for_display + + if path.startswith("models:/"): + from ._mlflow_identity import _redact_mlflow_error_for_display + + return _redact_mlflow_error_for_display(path) + if is_stream_url(path): + return f"stream://{redact_stream_url_for_display(path[9:])}" + if ( + is_cloud_url(path) + or is_cleartext_cloud_url(path) + or is_pytorch_hub_url(path) + or is_cleartext_pytorch_hub_url(path) + ): + return redact_url_for_display(path) + parsed = urlparse(path) + hostname = _normalize_hostname(parsed.hostname or "") + if ( + parsed.scheme in {"http", "https"} + and hostname + and "/artifactory/" in parsed.path + and ( + _is_jfrog_service_host(hostname) + or _is_local_jfrog_host(hostname) + or hostname in _get_configured_jfrog_hosts() + ) + ): + netloc = parsed.netloc + if "@" in netloc: + netloc = parsed.hostname or "" + try: + port = parsed.port + except ValueError: + port = None + if port is not None: + netloc = f"{netloc}:{port}" + netloc = f"@{netloc}" + return urlunparse((parsed.scheme, netloc, parsed.path, "", "", "")) + return _huggingface_classification_url(path) + + def _is_path_within_directory(path: str, directory: str) -> bool: try: normalized_path = os.path.normcase(os.path.realpath(path)) @@ -479,27 +548,39 @@ def _is_non_filesystem_identifier(path: str) -> bool: return False -def _redacted_component_identity( - path: str, - sha256: str = "", - bom_ref_state: _BomRefState | None = None, +def _component_identity( + path: str, sha256: str, bom_ref_state: _BomRefState | None = None, reported_path: str | None = None ) -> tuple[str, str]: - """Return credential-safe component name and bom-ref values for exported SBOMs.""" - safe_identifier = redact_source_identifier(path) - component_name = os.path.basename(safe_identifier) or safe_identifier - if safe_identifier == path: - base_reference = safe_identifier - else: - base_reference = redact_source_reference(path) - if sha256: - base_reference = f"{base_reference}#modelaudit-content-sha256-{sha256}" - - if bom_ref_state is None: - return component_name, base_reference - return component_name, bom_ref_state.allocate( - base_reference, - literal_reference=safe_identifier == path, - ) + """Preserve source identities and disambiguate bounded or expanded references.""" + identifier = serialize_source_identifier(path if reported_path is None else reported_path) + reference = identifier + if identifier != path and sha256: + reference = f"{reference}#modelaudit-content-sha256-{sha256}" + if bom_ref_state: + reference = bom_ref_state.allocate(reference, literal_reference=identifier == path) + return os.path.basename(identifier) or identifier, reference + + +def _source_order( + paths: Iterable[str], risk_score: Callable[[str], int], metadata_for: Callable[[str], Any] +) -> list[str]: + """Sort raw sources, retaining tie order when oversized identifiers are bounded.""" + + def order_key(path: str) -> tuple[str, bool, int, str]: + identifier = serialize_source_identifier(path) + if identifier == path: + return identifier, False, 0, "" + metadata = metadata_for(path) + if hasattr(metadata, "model_dump"): + metadata = metadata.model_dump(mode="python") + metadata = serialize_source_value(metadata or {}) + if isinstance(metadata, dict): + metadata.pop("risk_score", None) + metadata.pop("scan_timestamp", None) + encoded = json.dumps(metadata, sort_keys=True, separators=(",", ":"), default=repr) + return identifier, True, -risk_score(path), hashlib.sha256(encoded.encode()).hexdigest() + + return sorted(dict.fromkeys(paths), key=order_key) def _trusted_metadata_fallback_paths(assets: Iterable[Any] | None) -> set[str]: @@ -525,6 +606,7 @@ def _calculate_risk_score(path: str, issues: list[Issue]) -> int: """Calculate risk score for a file based on associated issues.""" score = 0 for issue in issues: + issue = finding_identity(issue) if issue.location == path: if issue.severity == IssueSeverity.CRITICAL: score += 5 @@ -539,6 +621,7 @@ def _calculate_legacy_risk_score(path: str, issues: Iterable[dict[str, Any]]) -> """Calculate the legacy dict-based risk score for a component.""" score = 0 for issue in issues: + issue = finding_identity(issue) if issue.get("location") != path: continue severity = issue.get("severity") @@ -551,47 +634,6 @@ def _calculate_legacy_risk_score(path: str, issues: Iterable[dict[str, Any]]) -> return min(score, 10) -def _stable_source_order( - paths: Iterable[str], - risk_score: Callable[[str], int], - metadata_identity: Callable[[str], str] | None = None, -) -> list[str]: - """Order sources so colliding redacted refs keep deterministic risk attribution.""" - grouped_paths: dict[str, list[str]] = {} - seen_paths: set[str] = set() - for path in paths: - if path in seen_paths: - continue - seen_paths.add(path) - reference = redact_source_reference(path) - grouped_paths.setdefault(reference, []).append(path) - - ordered_paths: list[str] = [] - for reference in sorted(grouped_paths): - ordered_paths.extend( - sorted( - grouped_paths[reference], - key=lambda path: ( - redact_source_identifier(path) != path, - -risk_score(path), - metadata_identity(path) if metadata_identity is not None else "", - ), - ) - ) - return ordered_paths - - -def _metadata_identity_sort_key(metadata: Any) -> str: - if hasattr(metadata, "model_dump"): - metadata = metadata.model_dump(mode="python") - safe_metadata = redact_source_value(metadata or {}) - if isinstance(safe_metadata, dict): - safe_metadata.pop("risk_score", None) - safe_metadata.pop("scan_timestamp", None) - serialized = json.dumps(safe_metadata, sort_keys=True, separators=(",", ":"), default=repr) - return hashlib.sha256(serialized.encode()).hexdigest() - - def _extract_license_expressions(metadata: FileMetadataModel) -> list[LicenseExpression]: """Extract license expressions from file metadata.""" license_expressions: list[LicenseExpression] = [] @@ -699,6 +741,8 @@ def _component_for_file_pydantic( relative_path: str | None = None, require_stable_root: bool = False, bom_ref_state: _BomRefState | None = None, + classification_path: str | None = None, + reported_path: str | None = None, ) -> Component: """Create a CycloneDX component from Pydantic models (type-safe version).""" size, sha256 = _resolve_component_size_and_sha256( @@ -715,7 +759,9 @@ def _component_for_file_pydantic( props = [Property(name="size", value=str(size))] # Calculate and add risk score - risk_score = _calculate_risk_score(path, issues) + if classification_path is None: + classification_path = _source_identity_path(path, metadata) + risk_score = _calculate_risk_score(classification_path, issues) props.append(Property(name="risk_score", value=str(risk_score))) # Add metadata-based properties if available @@ -725,8 +771,8 @@ def _component_for_file_pydantic( props.extend(_create_metadata_properties(metadata)) # Determine appropriate component type for CycloneDX v1.6 - component_type = _get_component_type(path, metadata.model_dump() if metadata else None) - component_name, bom_ref = _redacted_component_identity(path, sha256, bom_ref_state) + component_type = _get_component_type(classification_path, metadata.model_dump() if metadata else None) + component_name, bom_ref = _component_identity(path, sha256, bom_ref_state, reported_path) # Create the component component = Component( @@ -768,7 +814,8 @@ def _component_for_file( props = [Property(name="size", value=str(size))] # Compute risk score based on issues related to this file - score = _calculate_legacy_risk_score(path, issues) + classification_path = _source_identity_path(path, metadata) + score = _calculate_legacy_risk_score(classification_path, issues) props.append(Property(name="risk_score", value=str(score))) # Enhanced license handling @@ -858,8 +905,8 @@ def _component_for_file( props.append(Property(name="security:scanner_version", value=SCANNER_VERSION)) # Determine appropriate component type for CycloneDX v1.6 - component_type = _get_component_type(path, metadata if isinstance(metadata, dict) else None) - component_name, bom_ref = _redacted_component_identity(path, sha256, bom_ref_state) + component_type = _get_component_type(classification_path, metadata if isinstance(metadata, dict) else None) + component_name, bom_ref = _component_identity(path, sha256, bom_ref_state) component = Component( name=component_name, @@ -890,14 +937,12 @@ def generate_sbom(paths: Iterable[str], results: dict[str, Any] | Any) -> str: file_meta: dict[str, Any] = results.get("file_metadata", {}) trusted_metadata_paths = _trusted_metadata_fallback_paths(results.get("assets", [])) - ordered_paths = _stable_source_order( + ordered_paths = _source_order( paths, - lambda path: _calculate_legacy_risk_score(path, issues_dicts), - lambda path: _metadata_identity_sort_key(file_meta.get(path)), - ) - bom_ref_state = _BomRefState( - reserved={redact_source_reference(path) for path in ordered_paths if redact_source_identifier(path) == path} + lambda path: _calculate_legacy_risk_score(_source_identity_path(path, file_meta.get(path)), issues_dicts), + file_meta.get, ) + bom_ref_state = _BomRefState(reserved={path for path in ordered_paths if serialize_source_identifier(path) == path}) for input_path in ordered_paths: is_remote_identifier = _is_non_filesystem_identifier(input_path) if not is_remote_identifier and os.path.isdir(input_path): @@ -968,7 +1013,13 @@ def generate_sbom(paths: Iterable[str], results: dict[str, Any] | Any) -> str: return str(outputter.output_as_string(indent=2)) -def generate_sbom_pydantic(paths: Iterable[str], results: ModelAuditResultModel) -> str: +def generate_sbom_pydantic( + paths: Iterable[str], + results: ModelAuditResultModel, + *, + _classification_paths: dict[str, str] | None = None, + _reported_paths: dict[str, str] | None = None, +) -> str: """ Generate SBOM directly from Pydantic models (type-safe version). @@ -982,14 +1033,14 @@ def generate_sbom_pydantic(paths: Iterable[str], results: ModelAuditResultModel) file_metadata: dict[str, FileMetadataModel] = results.file_metadata or {} trusted_metadata_paths = _trusted_metadata_fallback_paths(results.assets) - ordered_paths = _stable_source_order( + ordered_paths = _source_order( paths, - lambda path: _calculate_risk_score(path, issues), - lambda path: _metadata_identity_sort_key(file_metadata.get(path)), - ) - bom_ref_state = _BomRefState( - reserved={redact_source_reference(path) for path in ordered_paths if redact_source_identifier(path) == path} + lambda path: _calculate_risk_score( + (_classification_paths or {}).get(path, _source_identity_path(path, file_metadata.get(path))), issues + ), + file_metadata.get, ) + bom_ref_state = _BomRefState(reserved={path for path in ordered_paths if serialize_source_identifier(path) == path}) for input_path in ordered_paths: is_remote_identifier = _is_non_filesystem_identifier(input_path) if not is_remote_identifier and os.path.isdir(input_path): @@ -1040,6 +1091,8 @@ def generate_sbom_pydantic(paths: Iterable[str], results: ModelAuditResultModel) relative_path=(None if single_scan_root is None else os.path.relpath(input_path, input_directory)), require_stable_root=require_stable_root, bom_ref_state=bom_ref_state, + classification_path=(_classification_paths or {}).get(input_path), + reported_path=(_reported_paths or {}).get(input_path), ) bom.components.add(component) finally: diff --git a/modelaudit/integrations/source_serialization.py b/modelaudit/integrations/source_serialization.py new file mode 100644 index 000000000..a0687245e --- /dev/null +++ b/modelaudit/integrations/source_serialization.py @@ -0,0 +1,106 @@ +"""Bounded conversion of report values without changing their evidence.""" + +import hashlib +from collections.abc import Callable +from typing import Any + +from pydantic import AnyUrl, BaseModel + +_MAX_DEPTH = 32 +_MAX_STRING_CHARS = 256 * 1024 + + +def serialize_source_identifier(value: str) -> str: + if len(value) <= _MAX_STRING_CHARS: + return value + digest = hashlib.sha256(value.encode("utf-8", errors="surrogatepass")).hexdigest() + preview = value[:256].encode("utf-8", errors="backslashreplace").decode("utf-8") + return f"modelaudit-source:{preview}..." + + +def serialize_source_text(value: str) -> str: + return value if len(value) <= _MAX_STRING_CHARS else "" + + +def serialize_source_value(value: Any, *, identifier_key: Callable[[str], str] | None = None) -> Any: + """Preserve report shapes and JSON-compatible keys, bounding recursive values.""" + return serialize_source_values([value], identifier_key=identifier_key)[0] + + +def serialize_source_values(values: list[Any], *, identifier_key: Callable[[str], str] | None = None) -> list[Any]: + """Share identifier allocation while retaining each value's depth budget.""" + identifiers: dict[str, str] = {} + reserved: set[str] = set() + + def reserve(text: str) -> str: + if len(text) <= _MAX_STRING_CHARS: + reserved.add(text) + return text + if text not in identifiers: + identifiers[text] = serialize_source_identifier(text) + return text + + # Materialize each model/key once; strings remain references until IDs are allocated. + converted = [_serialize(value, set(), 0, reserve) for value in values] + reserved_identifiers = ( + {identifier_key(text) for text in reserved} if identifiers and identifier_key is not None else set() + ) + for text in sorted(identifiers, key=identifiers.__getitem__): + base = candidate = identifiers[text] + occurrence = 1 + while candidate in reserved or ( + identifier_key is not None and identifier_key(candidate) in reserved_identifiers + ): + occurrence += 1 + candidate = f"{base}#{occurrence}" + reserved.add(candidate) + if identifier_key is not None: + reserved_identifiers.add(identifier_key(candidate)) + identifiers[text] = candidate + return [_serialize(value, set(), 0, lambda text: identifiers.get(text, text)) for value in converted] + + +def _serialize(value: Any, seen: set[int], depth: int, transform: Callable[[str], str]) -> Any: + if depth > _MAX_DEPTH: + return "" + if isinstance(value, BaseModel): + return _serialize(value.model_dump(mode="python"), seen, depth + 1, transform) + if isinstance(value, AnyUrl): + value = str(value) + if isinstance(value, (bytes, bytearray)): + try: + value = bytes(value).decode("utf-8") + except UnicodeDecodeError: + return "" + if isinstance(value, str): + return transform(value) + if not isinstance(value, (dict, list, tuple, set, frozenset)): + return value + if id(value) in seen: + return "" + seen.add(id(value)) + try: + if isinstance(value, dict): + result: dict[Any, Any] = {} + occurrences: dict[str, int] = {} + for key, item in value.items(): + key = _serialize(key, set(), 0, transform) + if not isinstance(key, (str, int, float, bool)) and key is not None: + key = transform(str(key)) + if key in result: + base_key = str(key) + occurrence = occurrences.get(base_key, 2) + candidate = f"{base_key}#modelaudit-redacted-key-{occurrence}" + while candidate in result: + occurrence += 1 + candidate = f"{base_key}#modelaudit-redacted-key-{occurrence}" + occurrences[base_key] = occurrence + 1 + key = candidate + result[key] = _serialize(item, seen, depth + 1, transform) + return result + items = [_serialize(item, seen, depth + 1, transform) for item in value] + if isinstance(value, tuple): + return tuple(items) + return sorted(items, key=repr) if isinstance(value, (set, frozenset)) else items + finally: + seen.remove(id(value)) diff --git a/modelaudit/models.py b/modelaudit/models.py index 2419d0d43..a2227e294 100644 --- a/modelaudit/models.py +++ b/modelaudit/models.py @@ -18,6 +18,7 @@ ScanResult, normalize_unclassified_scan_failure, ) +from .utils.helpers.finding_identity import finding_identity # We'll use forward references and rebuild models after imports @@ -809,6 +810,7 @@ def aggregate_scan_result_direct(self, scan_result: Any) -> None: why=issue.why, type=getattr(issue, "type", None), # Include type if available rule_code=getattr(issue, "rule_code", None), + **({"finding_identity": issue.finding_identity} if hasattr(issue, "finding_identity") else {}), ) ) @@ -825,6 +827,7 @@ def aggregate_scan_result_direct(self, scan_result: Any) -> None: severity=check.severity if check.severity else None, why=check.why, rule_code=getattr(check, "rule_code", None), + **({"finding_identity": check.finding_identity} if hasattr(check, "finding_identity") else {}), ) ) @@ -870,7 +873,8 @@ def deduplicate_issues(self) -> None: deduplicated_issues = [] for issue in self.issues: # Include location in the deduplication key to avoid hiding issues in different files - issue_key = (issue.message, issue.severity, issue.location or "") + identity = finding_identity(issue) + issue_key = (identity.message, identity.severity, identity.location or "") if issue_key not in seen_issues: seen_issues.add(issue_key) deduplicated_issues.append(issue) diff --git a/modelaudit/scanners/_catboost_evidence_redaction.py b/modelaudit/scanners/_catboost_evidence_redaction.py deleted file mode 100644 index a01b63efb..000000000 --- a/modelaudit/scanners/_catboost_evidence_redaction.py +++ /dev/null @@ -1,4221 +0,0 @@ -"""Helpers for storing scanner evidence without embedded secrets.""" - -from __future__ import annotations - -import ast -import io -import re -import shlex -import string -import tokenize -import unicodedata -from typing import Final, TypeAlias -from urllib.parse import SplitResult, parse_qsl, unquote, unquote_plus, urlencode, urlsplit, urlunsplit - -from ._evidence_redaction import ( - AUTHORIZATION_ALIAS_ASSIGNMENT_KEY as SHARED_AUTHORIZATION_ALIAS_ASSIGNMENT_KEY, -) -from ._evidence_redaction import ( - CAMEL_CASE_SENSITIVE_ASSIGNMENT_KEY as SHARED_CAMEL_CASE_SENSITIVE_ASSIGNMENT_KEY, -) -from ._evidence_redaction import ( - STANDALONE_SECRET_RE as SHARED_STANDALONE_SECRET_RE, -) - -REDACTED_EVIDENCE_VALUE: Final[str] = "" -REDACTED_URL_CREDENTIALS: Final[str] = "" -EVIDENCE_REDACTION_LOOKAHEAD_CHARS: Final[int] = 4096 -EVIDENCE_URL_LOOKAHEAD_CHARS: Final[int] = 64 * 1024 -CURL_COMMAND_SCAN_CHARS: Final[int] = EVIDENCE_URL_LOOKAHEAD_CHARS -MAX_CURL_EXECUTABLE_CANDIDATES: Final[int] = 64 -MAX_SUBPROCESS_CALL_CANDIDATES: Final[int] = 64 -MAX_COMPARISON_CANDIDATES: Final[int] = 64 -MAX_URL_QUERY_DECODE_PASSES: Final[int] = 8 -MAX_NESTED_URL_QUERY_DEPTH: Final[int] = 8 -MAX_EVALUATED_KEY_CHARS: Final[int] = 256 -MAX_KEY_EXPRESSION_CHARS: Final[int] = 300 -MAX_KEY_EXPRESSION_PARSE_ATTEMPTS: Final[int] = 64 -MAX_KEY_EXPRESSION_CANDIDATES: Final[int] = 8 -SENSITIVE_KEYWORD_SIGNALS: Final[tuple[str, ...]] = ( - "access", - "api", - "auth", - "cookie", - "credential", - "jwt", - "pass", - "private", - "pwd", - "sas", - "secret", - "session", - "sig", - "storage", - "token", -) -DICT_LOOKUP_MISSING: Final[object] = object() -DICT_LOOKUP_UNKNOWN: Final[object] = object() -NONE_COMPARABLE: Final[object] = object() -EvaluatedStringSequence: TypeAlias = list[str] | tuple[str, ...] | frozenset[str] -EvaluatedStringValue: TypeAlias = str | EvaluatedStringSequence -ComparableLiteral: TypeAlias = bool | int | EvaluatedStringValue -MembershipContainer: TypeAlias = str | list[object] | tuple[object, ...] | frozenset[object] - -URL_RE: Final[re.Pattern[str]] = re.compile(r"(?i)\b[a-z][a-z0-9+.-]*://[^\s\"'<>]+") -COMMAND_URL_CONTEXT_RE: Final[re.Pattern[str]] = re.compile(r"(?i)\b[a-z][a-z0-9+.-]*://[^\s\"']+") -PYTHON_STRING_PREFIX_RE: Final[str] = r"[rubf]{0,3}" -PYTHON_LITERAL_OPEN_RE: Final[str] = r"(?:\s*\(\s*)*" -PYTHON_STRING_LITERAL_FRAGMENT_RE: Final[str] = ( - rf"(?:{PYTHON_STRING_PREFIX_RE})(?:(?:\\*[\"']){{3}}[\s\S]*?(?:\\*[\"']){{3}}|" - r"(?:\\*[\"'])[\s\S]*?(?:\\*[\"']))" -) -PYTHON_STRING_LITERAL_DETECT_RE: Final[re.Pattern[str]] = re.compile(PYTHON_STRING_LITERAL_FRAGMENT_RE, re.IGNORECASE) -PYTHON_LITERAL_JOINED_FRAGMENT_RE: Final[str] = ( - rf"(?:(?:\s+|\s*\+\s*|\s*%\s*\(?\s*){PYTHON_STRING_LITERAL_FRAGMENT_RE}\s*\)?)" -) -PYTHON_RESIDUAL_LITERAL_OPERATOR_RE: Final[str] = r"(?:[\(\.,+*/%]|\b(?:or|and|if|else)\b)" -QUOTED_KEY_CONTENT_PATTERN: Final[str] = rf"(?:[^\\\"']|\\[\s\S]){{0,{MAX_KEY_EXPRESSION_CHARS}}}?" -SERIALIZED_BACKSLASH_ESCAPE_TOKEN: Final[str] = ( - r"\\+(?:u+005c|x5c|134|U0000005c|u\{0*5c\}|N\{(?:reverse solidus|backslash)\})" -) -SERIALIZED_BACKSLASH_ESCAPE_VALUE_RE: Final[re.Pattern[str]] = re.compile( - r"(?i)(?:u+005c|x5c|134|U0000005c|u\{0*5c\}|N\{(?:reverse solidus|backslash)\})" -) -SERIALIZED_QUOTE_ESCAPE_TOKEN: Final[str] = ( - r"(?:u+0022|u+0027|x22|x27|0?42|0?47|U00000022|U00000027|u\{0*22\}|u\{0*27\}|" - r"N\{(?:quotation mark|double quote|apostrophe|single quote)\})" -) -SERIALIZED_QUOTE_ESCAPE_RE: Final[re.Pattern[str]] = re.compile( - rf"(?i)((?:{SERIALIZED_BACKSLASH_ESCAPE_TOKEN})*)\\+({SERIALIZED_QUOTE_ESCAPE_TOKEN})" -) -SERIALIZED_BACKSLASH_ESCAPED_QUOTE_RE: Final[re.Pattern[str]] = re.compile( - rf"(?i)((?:{SERIALIZED_BACKSLASH_ESCAPE_TOKEN})+)(\\+)([\"'])" -) -SERIALIZED_SLASH_ESCAPE_RE: Final[re.Pattern[str]] = re.compile( - r"(?i)\\(?:x2f|u+002f|U0000002f|u\{0*2f\}|0?57|N\{solidus\})" -) -SERIALIZED_ASSIGNMENT_SEPARATOR_ESCAPE_RE: Final[re.Pattern[str]] = re.compile( - r"(?i)\\+(?:x(?:3a|3d)|u+(?:003a|003d)|U000000(?:3a|3d)|u\{0*(?:3a|3d)\}|0?(?:72|75)|" - r"N\{(?:colon|equals sign)\})" -) -SENSITIVE_QUERY_KEYS: Final[frozenset[str]] = frozenset( - { - "access_key", - "access-key", - "access_key_id", - "access-key-id", - "access_token", - "access-token", - "api_key", - "api-key", - "apikey", - "auth", - "aws_access_key_id", - "aws-secret-access-key", - "aws_secret_access_key", - "aws-session-token", - "aws_session_token", - "awsaccesskeyid", - "awssecretaccesskey", - "awssessiontoken", - "auth_token", - "auth-token", - "basic_auth", - "basic-auth", - "client_secret", - "client-secret", - "cookie", - "credential", - "jwt", - "passphrase", - "password", - "passwd", - "pwd", - "private_key", - "private-key", - "proxy-authorization", - "proxy_authorization", - "proxyauthorization", - "refresh_token", - "refresh-token", - "sas", - "secret", - "secret_key", - "secret-key", - "session_id", - "session-id", - "session_token", - "session-token", - "sessionid", - "sig", - "signature", - "token", - "x-amz-credential", - "x-amz-security-token", - "x-amz-signature", - } -) -COMPACT_SENSITIVE_QUERY_KEYS: Final[frozenset[str]] = frozenset( - re.sub(r"[._-]", "", key.lower()) for key in SENSITIVE_QUERY_KEYS -) -COMPACT_SENSITIVE_KEY_PREFIX: Final[str] = ( - r"(?:db|database|pg|postgres(?:ql)?|mysql|mariadb|mongo(?:db)?|redis|github|gitlab|bitbucket|slack|stripe|" - r"npm|pypi|docker|registry|huggingface|hf|openai|anthropic|azure|gcp|google)" -) -COMPACT_PREFIXED_SENSITIVE_ASSIGNMENT_KEY: Final[str] = ( - rf"(?:{COMPACT_SENSITIVE_KEY_PREFIX}(?:password|token|secret)|" - r"azurestoragekey|" - r"(?-i:[A-Z0-9]+(?:PASSWORD|TOKEN|SECRET)))" -) -SEPARATED_SENSITIVE_ASSIGNMENT_KEY: Final[str] = ( - r"(?:[a-z0-9]+[_.-])*" - rf"(?:{COMPACT_PREFIXED_SENSITIVE_ASSIGNMENT_KEY}|" - r"access[_.-]?key(?:[_.-]?id)?|access[_.-]?token|api[_.-]?key|apikey|" - r"aws(?:accesskeyid|secretaccesskey|sessiontoken)|auth[_.-]?token|client[_.-]?secret|" - r"auth|basic[_.-]?auth|cookie|credential|google[_.-]?access[_.-]?id|jwt|passphrase|password|passwd|" - r"private[_.-]?key|pwd|" - r"refresh[_.-]?token|sas|secret|" - r"secret[_.-]?key|session[_.-]?(?:id|token)|sessionid|signature|sig|storage[_.-]?key|token)" - r"(?:s|[0-9]+|[_.-]?values?)?" -) -CAMEL_CASE_SENSITIVE_NEAR_MATCH_SUFFIX: Final[str] = r"(?:Algorithm|Cache|Count|Format|Hint|Ingredient|Policy)" -CATBOOST_CAMEL_CASE_SENSITIVE_ASSIGNMENT_KEY: Final[str] = ( - rf"(?![A-Za-z0-9]*{CAMEL_CASE_SENSITIVE_NEAR_MATCH_SUFFIX}\b)" - rf"{SHARED_CAMEL_CASE_SENSITIVE_ASSIGNMENT_KEY}" -) -SENSITIVE_ASSIGNMENT_KEY: Final[str] = ( - rf"(?:{SEPARATED_SENSITIVE_ASSIGNMENT_KEY}|(?-i:{CATBOOST_CAMEL_CASE_SENSITIVE_ASSIGNMENT_KEY}))" -) -AUTHORIZATION_KEY_PATTERN: Final[str] = SHARED_AUTHORIZATION_ALIAS_ASSIGNMENT_KEY -ASSIGNMENT_SEPARATOR: Final[str] = r"(?::=|\*\*=|//=|<<=|>>=|[+\-*/%@&|^]=|[:=](?!=))" -ASSIGNMENT_SEPARATOR_RE: Final[re.Pattern[str]] = re.compile(rf"(?i)\s*{ASSIGNMENT_SEPARATOR}\s*") -KNOWN_AUTHORIZATION_SCHEME_PATTERN: Final[str] = ( - r"(?:bearer|basic|digest|token|negotiate|ntlm|aws4-hmac-sha256|foo\.bar)" -) -COMPOUND_AUTHORIZATION_SCHEME_PATTERN: Final[str] = r"(?:digest|aws4-hmac-sha256)" -AUTHORIZATION_SCHEME_PATTERN: Final[str] = rf"(?:{KNOWN_AUTHORIZATION_SCHEME_PATTERN}\s+)?" -SENSITIVE_ASSIGNMENT_KEY_RE: Final[re.Pattern[str]] = re.compile(rf"(?i)^{SENSITIVE_ASSIGNMENT_KEY}$") -COMPARISON_OPERATOR_PATTERN: Final[str] = ( - r"(?:===|!==|==|!=|>=|<=|<>|" - r"(?])<(?![<=>]|redacted>|credentials-redacted>)|" - r"(?])(?(?![=>])|" - r"(?\w.])" -) -COMMAND_SECRET_LONG_OPTION_NAME_PATTERN: Final[str] = ( - r"(?:account[_-]?key|storage[_-]?key|federated[_-]?token|cookie|password|passwd|passphrase|pass|" - r"proxy-password|proxy-passphrase|proxy-pass|" - r"proxy-tls-?password|tls-?password|ftp-account|ftp-password|http-password|oauth2-bearer|" - r"client[_-]?secret|api[_-]?key|token|secret)" -) -COMMAND_SECRET_LONG_OPTION_PATTERN: Final[str] = rf"--{COMMAND_SECRET_LONG_OPTION_NAME_PATTERN}" -COMMAND_SECRET_OPTION_PREFIX_PATTERN: Final[str] = ( - rf"(?:(?(?\$?\"(?:\\.|[^\"\\])*\"|\$?'(?:\\.|[^'\\])*'|[^\s;&|)]+)" -) -COMMAND_SHELL_SENSITIVE_ASSIGNMENT_RE: Final[re.Pattern[str]] = re.compile( - rf"(?is)(?P\b(?:{SENSITIVE_ASSIGNMENT_KEY})\s*=\s*)" - r"(?P\$?\"(?:\\.|[^\"\\])*\"|\$?'(?:\\.|[^'\\])*'|(?!\\+[\"'])[^\s\"';&|)]+)" -) -COMMAND_SECRET_OPTION_SERIALIZED_QUOTED_RE: Final[re.Pattern[str]] = re.compile( - rf"(?is)(?P