diff --git a/.github/secret_scanning.yml b/.github/secret_scanning.yml new file mode 100644 index 0000000..fcf2077 --- /dev/null +++ b/.github/secret_scanning.yml @@ -0,0 +1,16 @@ +# Paths excluded from secret scanning alerts. +# +# Scope this as narrowly as possible. An entry here means a real credential +# committed to that path would not raise an alert, so it is only appropriate +# for a file whose contents are credential-shaped by definition. +# +# The redaction tests assert that values in published credential formats are +# masked, which requires holding examples of those formats. They are assembled +# from a prefix and a body at runtime so no literal appears in the file, and +# this entry covers anything that slips past that. +# +# Note: this suppresses alerts only. Push protection is evaluated separately +# and ignores this file, so a credential-shaped literal will still block a +# push; assembling the value at runtime is what avoids that. +paths-ignore: + - "tests/test_secret_redaction.py" diff --git a/CHANGELOG.md b/CHANGELOG.md index a57e50c..317ce7c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,42 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +### Fixed +- **A finding's snippet no longer reproduces the value it reports.** A SAST + finding's `codeSnippet` is the source line the rule matched. For nearly every + rule that line is the code the finding is about; for the hardcoded-credential + rules it contains the credential, so the finding carried the value into + `.socket.facts.json`, the uploaded facts and the configured notifiers. + Snippets for those rules now keep the assignment target, the syntax, the file + and the line, and mask the literal's contents. This covers 20 rules across all + fifteen bundled language rule sets, not only the Python and JavaScript ones: + `*-hardcoded-secret(s)`, `*-hardcoded-credentials`, + `*-hardcoded-password-default`, `*-default-credentials`, + `*-plain-text-password`, `*-weak-jwt-secret` and `*-empty-password`. Rules + whose match is not a credential keep their snippets verbatim. +- Every snippet, dataflow-trace step and detailed report, whatever rule produced + it, is now masked of values matching a well-known credential format: AWS key + IDs, GitHub tokens, Stripe keys, Slack tokens, Google API keys, npm and PyPI + tokens, JWTs, PEM private key bodies, and credentials in a URL authority. A + rule unrelated to secrets can still match a line that carries one. +- TruffleHog's `redactedValue` kept the first and last four characters of any + value longer than eight, which left most of a short password readable. Values + under sixteen characters are now masked in full. +- TruffleHog no longer scans the facts file the run writes. That file lands + inside the scan target, so a previous run's output was on disk during the walk + and its contents were reported as findings of their own, pointing at the + output file rather than the source line. + +### Changed +- `load_explicit_env_config` builds its "API key sources detected" debug line by + iterating a tuple of variable names rather than a dict of presence booleans. + The line is unchanged, including the exclusion of an exported-but-empty + variable. + +### Added +- A `redact` rule-metadata key. Set it on a custom SAST rule to mark the match + as a credential, or to opt a rule out; without it, the rule name decides. + ## [3.3.0] - 2026-09-15 Small release pairing a CLI parity addition with a notification fix. The fix diff --git a/docs/parameters.md b/docs/parameters.md index 028cf40..6e011cf 100644 --- a/docs/parameters.md +++ b/docs/parameters.md @@ -467,6 +467,37 @@ Custom rule file notes: - Files ending in `.test.yml` or `.test.yaml` are ignored. - Rules without `languages` are skipped. +#### Masking a custom rule's matched line + +A finding's `codeSnippet` is the source line the rule matched. For most rules +that line is vulnerable code and is shown as written; for a rule that matches a +hardcoded credential, the line contains the credential, so Socket Basics masks +the string literals in it before the finding is written to the facts file, +uploaded, or sent to a notifier. The file path and line number are kept either +way. + +Bundled rules are recognized by name (`*-hardcoded-secret`, +`*-hardcoded-credentials`, `*-hardcoded-password`, `*-default-credentials`, +`*-plain-text-password`, `*-weak-jwt-secret`). A custom rule can say so +directly with a `redact` metadata key, which also works to opt a rule out: + +```yaml +rules: + - id: acme-internal-token + message: "Internal service token committed to source" + severity: HIGH + languages: [python] + pattern: $VAR = "acme_tok_..." + metadata: + redact: true +``` + +Independently of this setting, every snippet is scrubbed of values matching a +well-known credential format (AWS key IDs, GitHub tokens, Stripe keys, Slack +tokens, Google API keys, npm and PyPI tokens, JWTs, PEM private key bodies, and +credentials embedded in a URL), so a rule unrelated to secrets that matches a +line carrying one does not pass it through. + ### Language-Specific Rule Configuration For each language, you can enable or disable specific rules: diff --git a/socket_basics/core/config.py b/socket_basics/core/config.py index 84a877c..6b8c0da 100644 --- a/socket_basics/core/config.py +++ b/socket_basics/core/config.py @@ -1121,13 +1121,15 @@ def load_explicit_env_config() -> Dict[str, Any]: logger = logging.getLogger(__name__) config = {} - # Log which API key sources are available for debugging - api_key_sources = { - 'SOCKET_SECURITY_API_KEY': bool(os.environ.get('SOCKET_SECURITY_API_KEY')), - 'SOCKET_SECURITY_API_TOKEN': bool(os.environ.get('SOCKET_SECURITY_API_TOKEN')), - 'INPUT_SOCKET_SECURITY_API_KEY': bool(os.environ.get('INPUT_SOCKET_SECURITY_API_KEY')), - } - found_sources = [k for k, v in api_key_sources.items() if v] + # Log which API key sources are available for debugging. The list holds the + # variable names, which come from the tuple below; a variable's value is + # only ever tested for emptiness and is never carried into the log line. + api_key_env_vars = ( + 'SOCKET_SECURITY_API_KEY', + 'SOCKET_SECURITY_API_TOKEN', + 'INPUT_SOCKET_SECURITY_API_KEY', + ) + found_sources = [name for name in api_key_env_vars if os.environ.get(name)] if found_sources: logger.debug(f"API key sources detected: {', '.join(found_sources)}") diff --git a/socket_basics/core/connector/opengrep/__init__.py b/socket_basics/core/connector/opengrep/__init__.py index 30d4889..80f8630 100644 --- a/socket_basics/core/connector/opengrep/__init__.py +++ b/socket_basics/core/connector/opengrep/__init__.py @@ -23,6 +23,12 @@ from . import github_pr, slack, ms_teams, ms_sentinel, sumologic, console, jira, webhook, json_notifier from .custom_rules import CustomRulesBuilder from .cwe_catalog import CWE_CATALOG +from ...utils.redaction import ( + is_credential_finding, + redact_dataflow_trace, + redact_message, + redact_snippet, +) # Import shared formatters from ...formatters import get_all_formatters @@ -407,6 +413,14 @@ def _convert_to_socket_facts(self, raw_results: Any) -> Dict[str, Any]: severity = ((r.get('extra') or {}).get('severity') or r.get('severity') or '') severity_norm = str(severity).lower() if severity is not None else '' message = (r.get('extra') or {}).get('message') or r.get('message') or '' + credential_finding = is_credential_finding( + check_id, (r.get('extra') or {}).get('metadata') or {} + ) + message = redact_message( + message, + (r.get('extra') or {}).get('metavars') or {}, + credential_finding=credential_finding, + ) start = (r.get('start') or {}).get('line') end = (r.get('end') or {}).get('line') @@ -423,7 +437,15 @@ def _convert_to_socket_facts(self, raw_results: Any) -> Dict[str, Any]: # default if unknown sev_label = 'medium' + # Mask here, not at the destinations: everything below + # reads `code_snippet`, and the props it lands in are + # read in turn by the facts file, the upload and every + # notifier. See socket_basics.core.utils.redaction. code_snippet = (r.get('extra') or {}).get('lines') or (r.get('extra') or {}).get('snippet') or '' + code_snippet = redact_snippet( + code_snippet, + credential_finding=credential_finding, + ) alert = { 'title': check_id, @@ -527,7 +549,11 @@ def _fmt_trace_loc(trace_item): if _intermediates: _trace_data['intermediates'] = [_fmt_trace_loc(v) for v in _intermediates] - alert['props']['dataflowTrace'] = _trace_data + # Trace steps quote source lines the same way the + # snippet does, so they get the same treatment. + alert['props']['dataflowTrace'] = redact_dataflow_trace( + _trace_data, credential_finding + ) except Exception: pass # Skip trace if structure is unexpected diff --git a/socket_basics/core/connector/trufflehog/__init__.py b/socket_basics/core/connector/trufflehog/__init__.py index 715668e..251ec37 100644 --- a/socket_basics/core/connector/trufflehog/__init__.py +++ b/socket_basics/core/connector/trufflehog/__init__.py @@ -19,6 +19,7 @@ # coerce_bool lives in the config layer because the environment loader, a # Socket dashboard config, and a JSON config each deliver booleans differently. from ...config import coerce_bool +from ...utils.redaction import mask_value # Import individual notifier modules from . import github_pr, slack, ms_teams, ms_sentinel, sumologic, console, jira, webhook, json_notifier @@ -169,6 +170,31 @@ def _build_exclude_patterns(self, exclude_dirs: Any) -> List[str]: return patterns + def _output_file_patterns(self) -> List[str]: + """Exclude the facts file this run writes into the scanned workspace. + + The facts file lands inside the scan target, so a previous run's output + is on disk by the time TruffleHog walks the tree. Whatever another + scanner recorded there gets re-detected as a finding of its own, + pointing at the output file instead of the source line. Exclude both + the file and the temporary name it is staged under. + """ + output_name = self.config.get('output', '.socket.facts.json') or '.socket.facts.json' + output_dir = getattr(self.config, 'output_dir', None) or self._workspace_root() + if not output_dir: + return [] + try: + output_path = Path( + self._absolute_scan_target(Path(output_dir) / output_name) + ) + except (TypeError, ValueError): + return [] + + patterns = [] + for candidate in (output_path, output_path.with_name(output_path.name + '.tmp')): + patterns.append(rf'^{self._path_regex(str(candidate))}$') + return patterns + def _write_exclude_file(self, exclude_dirs: Any) -> str | None: """Write exclude regexes to a temporary file for TruffleHog.""" patterns = self._build_exclude_patterns(exclude_dirs) @@ -304,6 +330,8 @@ def scan(self) -> Dict[str, Any]: exclude_dirs = self.config.get('trufflehog_exclude_dir', '') if exclude_dirs: exclude_patterns = self._build_exclude_patterns(exclude_dirs) + exclude_patterns.extend(self._output_file_patterns()) + if exclude_patterns: logger.debug("TruffleHog exclude patterns: %s", exclude_patterns) exclude_file_path = self._write_exclude_patterns(exclude_patterns) if exclude_file_path: @@ -546,10 +574,11 @@ def _create_alert(self, finding: Dict[str, Any]) -> Dict[str, Any]: file_path = self._workspace_relative_path(file_path) - # Redact the actual secret - raw_secret = finding.get('Raw', '') - redacted_secret = raw_secret[:4] + '*' * (len(raw_secret) - 8) + raw_secret[-4:] if len(raw_secret) > 8 else '*' * len(raw_secret) - + # TruffleHog reports the match verbatim in `Raw` and leaves its own + # `Redacted` field empty for most detectors, so neither field can be + # passed through and the masking has to happen here. + redacted_secret = mask_value(finding.get('Raw', '')) + markdown_content = f"""## Secret Detected: {detector_name} ### Detection Details diff --git a/socket_basics/core/utils/redaction.py b/socket_basics/core/utils/redaction.py new file mode 100644 index 0000000..6f69474 --- /dev/null +++ b/socket_basics/core/utils/redaction.py @@ -0,0 +1,303 @@ +"""Masking for finding fields that would otherwise reproduce a credential. + +A finding needs to say *where* a credential is, not *what* it is. Everything a +connector puts on an alert travels further than the scanned checkout: it is +written to the facts file in the workspace, uploaded to Socket, rendered in the +dashboard, and pasted into whichever notifiers are configured. Anything copied +verbatim out of the source line is therefore copied into all of those places, +so the copy has to be masked at the point the alert is built rather than at +each destination. + +Two passes are available: + +``scrub_tokens`` + Always safe to run. Masks only strings matching a well-known credential + format (AWS key IDs, GitHub tokens, PEM private key bodies, and so on), + which are specific enough that a match is not a guess. + +``redact_literals`` + For findings whose whole subject is a hardcoded credential. Masks the body + of every string literal on the line, keeping the assignment target and the + surrounding syntax so the finding is still readable. + +``redact_snippet`` composes them, ``redact_message`` removes values interpolated +into rule messages, and ``is_credential_finding`` decides which rules get the +second pass. +""" + +import re +from typing import Any, Mapping + +__all__ = [ + "mask_value", + "scrub_tokens", + "redact_literals", + "redact_snippet", + "redact_message", + "redact_dataflow_trace", + "is_credential_finding", +] + +# Below this length the revealed head and tail are a large enough fraction of +# the value to narrow it down, so short values are masked in full. +_MIN_LENGTH_FOR_PARTIAL_REVEAL = 16 +_DEFAULT_REVEAL = 4 + +# At or above this length a bound metavariable value is specific enough that +# replacing it anywhere in a message is safe; below it, the replace is anchored +# to non-word boundaries instead. +_MIN_STANDALONE_METAVAR_LENGTH = 8 + + +def mask_value(value: Any, reveal: int = _DEFAULT_REVEAL, + min_length: int = _MIN_LENGTH_FOR_PARTIAL_REVEAL) -> str: + """Mask a credential, keeping enough shape to tell two findings apart. + + Values at least ``min_length`` long keep ``reveal`` leading and trailing + characters; shorter ones are masked completely. The asterisk run matches + the original length so the result still lines up with the source. + """ + text = value if isinstance(value, str) else str(value or '') + if not text: + return text + if len(text) < min_length or len(text) <= reveal * 2: + return '*' * len(text) + return f"{text[:reveal]}{'*' * (len(text) - 2 * reveal)}{text[-reveal:]}" + + +# Formats distinctive enough that a match is a credential rather than a string +# that happens to look like one. Each pattern captures the secret in group 1; +# fixed vendor prefixes stay outside the group because they identify the key +# type without disclosing anything. +_TOKEN_PATTERNS = ( + # AWS access key IDs (AKIA/ASIA/ABIA/ACCA/A3T + 16 chars). + re.compile(r'\b((?:A3T[A-Z0-9]|AKIA|ABIA|ACCA|ASIA)[A-Z0-9]{16})\b'), + # GitHub personal access, OAuth, user-to-server, server-to-server and + # refresh tokens. + re.compile(r'\b(gh[pousr]_[A-Za-z0-9]{20,255})\b'), + re.compile(r'\b(github_pat_[A-Za-z0-9_]{20,255})\b'), + # Stripe secret, restricted and publishable keys. + re.compile(r'\b((?:sk|rk|pk)_(?:live|test)_[A-Za-z0-9]{10,})\b'), + # Slack bot/user/app/refresh tokens and legacy workspace tokens. + re.compile(r'\b(xox[abeoprs]-[A-Za-z0-9-]{10,})\b'), + # Google API keys. + re.compile(r'\b(AIza[A-Za-z0-9_\-]{35})\b'), + # OpenAI-style project and user keys. + re.compile(r'\b(sk-(?:proj-)?[A-Za-z0-9_\-]{20,})\b'), + # npm and PyPI upload tokens. + re.compile(r'\b(npm_[A-Za-z0-9]{36})\b'), + re.compile(r'\b(pypi-[A-Za-z0-9_\-]{16,})\b'), + # Twilio account SIDs and API keys. + re.compile(r'\b((?:AC|SK)[0-9a-fA-F]{32})\b'), + # SendGrid. + re.compile(r'\b(SG\.[A-Za-z0-9_\-]{16,}\.[A-Za-z0-9_\-]{16,})\b'), + # JSON Web Tokens: the payload segment carries the claims. + re.compile(r'\b(eyJ[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]*)'), + # Credentials embedded in a URL's authority section. + re.compile(r'://[^\s:/@]+:([^\s/@]+)@'), +) + +# PEM private keys are masked as a block: the base64 body is the key material +# and its line structure carries nothing worth keeping. +_PEM_BLOCK = re.compile( + r'(-----BEGIN [A-Z ]*PRIVATE KEY-----)(.*?)(-----END [A-Z ]*PRIVATE KEY-----)', + re.DOTALL, +) + +# Quoted string literals, including escaped quotes. Covers the single, double +# and backtick forms the bundled rules match across languages. +_STRING_LITERAL = re.compile( + r"""(?P["'`])(?P(?:\\.|(?!(?P=quote))[^\\])*)(?P=quote)""", + re.DOTALL, +) + +# Fallback for unquoted forms such as ``password: hunter2`` in config-style +# sources, used only when a credential finding has no string literal to mask. +# +# The operator alternation is what keeps a quoted value out of this branch. A +# bare ``[=:]`` stops on the first character of ``:=`` or ``==`` and leaves the +# rest of the operator at the head of the value, which no longer looks quoted, +# so a Go short declaration or a comparison would be starred out whole instead +# of going to the literal pass. ``=`` and ``:`` are matched only where they are +# not part of a longer operator, and a comparison assigns nothing, so it does +# not match here at all. +_UNQUOTED_ASSIGNMENT = re.compile( + r'^(?P[^=:]*(?::=|=(?!=)|:(?!:))\s*)(?P\S.*?)(?P\s*)$' +) + +# Rule-name fragments whose finding *is* the credential. ``hardcoded-ip`` and +# the password-policy rules deliberately do not appear: their snippets are +# logic, and masking them would remove the reason the finding was raised. +_CREDENTIAL_RULE_FRAGMENTS = ( + 'hardcoded-secret', + 'hardcoded-credential', + 'hardcoded-password', + 'hardcoded-key', + 'hardcoded-token', + 'default-credentials', + 'plain-text-password', + 'empty-password', + 'weak-jwt-secret', + 'private-key', + 'api-key', +) + + +def _mask_match(match: 're.Match[str]') -> str: + """Replace a pattern's captured secret in place, leaving its prefix intact.""" + whole = match.group(0) + secret = match.group(1) + if not secret: + return whole + secret_start, secret_end = match.span(1) + match_start = match.start(0) + relative_start = secret_start - match_start + relative_end = secret_end - match_start + return f"{whole[:relative_start]}{mask_value(secret)}{whole[relative_end:]}" + + +def scrub_tokens(text: Any) -> str: + """Mask well-known credential formats anywhere in ``text``.""" + if not isinstance(text, str) or not text: + return text if isinstance(text, str) else '' + + scrubbed = _PEM_BLOCK.sub( + lambda m: f"{m.group(1)}\n{'*' * 32}\n{m.group(3)}", text + ) + for pattern in _TOKEN_PATTERNS: + scrubbed = pattern.sub(_mask_match, scrubbed) + return scrubbed + + +def redact_literals(text: Any) -> str: + """Mask the body of every string literal, keeping the surrounding syntax. + + ``API_KEY = "sk_live_abc123"`` becomes ``API_KEY = "****************"``: + the name, the operator and the line all survive, which is what makes the + finding actionable, while the value does not. + """ + if not isinstance(text, str) or not text: + return text if isinstance(text, str) else '' + + def _mask_literal(match: 're.Match[str]') -> str: + body = match.group('body') + if not body: + return match.group(0) + quote = match.group('quote') + return f'{quote}{mask_value(body)}{quote}' + + # Mask unquoted assignments line by line before processing literals. A + # quoted value is left for the literal pass, while an unquoted value is + # masked even if another line or a trailing comment contains quoted text. + masked_lines = [] + for line in text.split('\n'): + match = _UNQUOTED_ASSIGNMENT.match(line) + body = match.group('body') if match else '' + if match and not body.lstrip().startswith(('"', "'", '`')): + # If quoted text appears later in an unquoted value, mask the whole + # body. Measuring that combined text could otherwise make a short + # credential eligible for a partial reveal. + masked_body = ( + '*' * len(body) if _STRING_LITERAL.search(body) else mask_value(body) + ) + masked_lines.append( + f"{match.group('head')}{masked_body}{match.group('tail')}" + ) + else: + masked_lines.append(line) + return _STRING_LITERAL.sub(_mask_literal, '\n'.join(masked_lines)) + + +def is_credential_finding(rule_id: Any, metadata: Mapping[str, Any] | None = None) -> bool: + """Report whether a rule's match is itself a credential. + + A rule can state this directly with a ``redact`` metadata key, which is how + custom rules opt in or out; otherwise the rule name decides. + """ + if isinstance(metadata, Mapping) and 'redact' in metadata: + declared = metadata.get('redact') + if isinstance(declared, str): + return declared.strip().lower() in ('1', 'true', 'yes', 'on') + return bool(declared) + + name = str(rule_id or '').lower() + return any(fragment in name for fragment in _CREDENTIAL_RULE_FRAGMENTS) + + +def redact_snippet(text: Any, credential_finding: bool = False) -> str: + """Mask a snippet before it is attached to an alert. + + ``credential_finding`` adds the string-literal pass on top of the token + scrub that every snippet receives. + """ + scrubbed = scrub_tokens(text) + if credential_finding: + scrubbed = redact_literals(scrubbed) + return scrubbed + + +def redact_message(text: Any, metavars: Any = None, + credential_finding: bool = False) -> str: + """Mask credentials interpolated into a scanner rule's message. + + OpenGrep expands metavariables before returning a result. For credential + findings, mask every expanded metavariable that appears in the message; + the connector cannot reliably identify which metavariable held the secret. + Well-known token formats are scrubbed from every message independently. + """ + redacted = scrub_tokens(text) + if not credential_finding or not isinstance(metavars, Mapping): + return redacted + + values = set() + for details in metavars.values(): + if not isinstance(details, Mapping): + continue + value = details.get('abstract_content') + if isinstance(value, str) and value: + values.add(value) + + for value in sorted(values, key=len, reverse=True): + if len(value) >= _MIN_STANDALONE_METAVAR_LENGTH: + # Long enough to be specific to itself. + redacted = redacted.replace(value, mask_value(value)) + else: + # A short bound value is also an ordinary substring, and an + # unanchored replace masks every occurrence rather than the one + # that is the credential: "a" turns "secret a is bad" into + # "secret * is b*d". Requiring a non-word character on each side + # keeps the credential masked without touching words that merely + # contain it. Skipping short values outright is not an option -- + # a short credential still has to be masked. + redacted = re.sub( + rf'(? Any: + """Mask the code fragments carried by a taint-mode dataflow trace. + + Each step quotes a source line, so a step gets the same treatment the + snippet does: the token scrub always, and the string-literal pass when the + rule's match is a credential. A taint rule reaches this for a credential + only by declaring ``redact`` in its metadata, since none of the bundled + credential rules are taint-mode, but the trace must not be the one field + that keeps the value when one does. + """ + if not isinstance(trace, dict): + return trace + + def _mask(step: Any) -> None: + if isinstance(step, dict) and step.get('content'): + step['content'] = redact_snippet(step['content'], credential_finding) + + for key in ('source', 'sink'): + _mask(trace.get(key)) + intermediates = trace.get('intermediates') + if isinstance(intermediates, list): + for step in intermediates: + _mask(step) + return trace diff --git a/tests/test_changed_files_scope.py b/tests/test_changed_files_scope.py index 0f268e3..4322c9a 100644 --- a/tests/test_changed_files_scope.py +++ b/tests/test_changed_files_scope.py @@ -943,16 +943,19 @@ def record_run(cmd, *args, **kwargs): TruffleHogScanner(cfg).scan() assert staged_calls == [] - assert invocations == [ - [ - "trufflehog", - "filesystem", - "--json", - "--include-detectors=all", - "--fail-on-scan-errors", - "--results=verified,unknown", - str(pr_repo), - ] + assert len(invocations) == 1 + # --exclude-paths always carries the facts file and is not part of the + # scope this test covers. + command = invocations[0] + exclude_index = command.index("--exclude-paths") + assert command[:exclude_index] + command[exclude_index + 2:] == [ + "trufflehog", + "filesystem", + "--json", + "--include-detectors=all", + "--fail-on-scan-errors", + "--results=verified,unknown", + str(pr_repo), ] def test_staged_fallback_still_runs_when_no_scope_was_requested(self, pr_repo, monkeypatch): diff --git a/tests/test_config_source.py b/tests/test_config_source.py index 4c02455..9ded38b 100644 --- a/tests/test_config_source.py +++ b/tests/test_config_source.py @@ -1,6 +1,6 @@ import logging -from socket_basics.core.config import Config +from socket_basics.core.config import Config, load_explicit_env_config def test_config_logs_default_environment_source(caplog, tmp_path): @@ -17,3 +17,31 @@ def test_config_logs_named_source(caplog, tmp_path): Config({"workspace": str(tmp_path), "_config_source": "api"}) assert "Configuration loaded from: Socket dashboard (API)" in caplog.text + + +def test_api_key_source_log_names_variables_not_values(caplog, monkeypatch): + """The debug line reports which variables are set, never what is in them.""" + caplog.set_level(logging.DEBUG, logger="socket_basics.core.config") + monkeypatch.setenv("SOCKET_SECURITY_API_KEY", "first-value-not-for-logs") + monkeypatch.setenv("INPUT_SOCKET_SECURITY_API_KEY", "second-value-not-for-logs") + monkeypatch.delenv("SOCKET_SECURITY_API_TOKEN", raising=False) + + load_explicit_env_config() + + assert "SOCKET_SECURITY_API_KEY" in caplog.text + assert "INPUT_SOCKET_SECURITY_API_KEY" in caplog.text + assert "first-value-not-for-logs" not in caplog.text + assert "second-value-not-for-logs" not in caplog.text + + +def test_an_empty_api_key_variable_is_not_reported_as_a_source(caplog, monkeypatch): + # An exported-but-empty variable is not a configured key, so it must not + # show up as one. + caplog.set_level(logging.DEBUG, logger="socket_basics.core.config") + monkeypatch.setenv("SOCKET_SECURITY_API_KEY", "") + monkeypatch.setenv("SOCKET_SECURITY_API_TOKEN", "a-real-looking-token") + monkeypatch.delenv("INPUT_SOCKET_SECURITY_API_KEY", raising=False) + + load_explicit_env_config() + + assert "API key sources detected: SOCKET_SECURITY_API_TOKEN" in caplog.text diff --git a/tests/test_secret_redaction.py b/tests/test_secret_redaction.py new file mode 100644 index 0000000..2862897 --- /dev/null +++ b/tests/test_secret_redaction.py @@ -0,0 +1,480 @@ +"""An alert must locate a credential without reproducing it. + +Every field a connector puts on an alert is written to the facts file in the +scanned workspace, uploaded to Socket, rendered in the dashboard, and forwarded +to the configured notifiers. A snippet copied verbatim from the source line +therefore ends up in all of those places, which for a hardcoded-credential +finding means the credential itself does. + +Masking happens where the alert is built, so these tests assert on the alert +rather than on any single destination: the notifiers all read +``props.codeSnippet`` and inherit whatever is there. +""" + +import glob +import json +import re +from pathlib import Path + +import pytest +import yaml + +from socket_basics.core.utils.redaction import ( + is_credential_finding, + mask_value, + redact_dataflow_trace, + redact_literals, + redact_message, + redact_snippet, + scrub_tokens, +) + +RULES_DIR = Path(__file__).resolve().parent.parent / "socket_basics" / "rules" + + +def _sample(prefix: str, body: str) -> str: + """Assemble a stand-in credential from its prefix and body. + + The values below are not real -- each is a published example or a + syntactically valid value of the right shape. They are still built at + runtime rather than written as literals, because the formats under test are + exactly the ones a scanner walking this repository looks for, and a literal + here would be reported as a finding in its own right. Splitting the string + keeps that quiet without weakening what the test asserts. + """ + return prefix + body + + +AWS_KEY_ID = _sample("AKIA", "IOSFODNN7EXAMPLE") +GITHUB_TOKEN = _sample("ghp_", "16C7e42F292c6912E7710c838347Ae178B4a") +GITHUB_PAT = _sample("github_pat_", "11ABCDEFG0abcdefghijkl_mnopqrstuvwxyz0123456789") +STRIPE_KEY = _sample("sk_live_", "51QwErTyUiOpAsDfGhJkLzXc") +SLACK_TOKEN = _sample("xoxb-", "123456789012-1234567890123-AbCdEfGhIjKlMnOpQrStUvWx") +GOOGLE_KEY = _sample("AIza", "SyD-1234567890abcdefghijklmnopqrstu") +NPM_TOKEN = _sample("npm_", "abcdefghijklmnopqrstuvwxyz0123456789") +PYPI_TOKEN = _sample("pypi-", "AgEIcHlwaS5vcmcCJDAwMDAwMDAw") +JWT = _sample( + "eyJhbGciOiJIUzI1NiJ9.", + "eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIn0" + ".dBjftJeZ4CVPmB92K27uhbUJU1p1r_wW1gFWFOEjXk", +) + +SAMPLE_TOKENS = [ + AWS_KEY_ID, + GITHUB_TOKEN, + GITHUB_PAT, + STRIPE_KEY, + SLACK_TOKEN, + GOOGLE_KEY, + NPM_TOKEN, + PYPI_TOKEN, +] + + +class TestMaskValue: + def test_long_values_keep_an_identifying_head_and_tail(self): + masked = mask_value(STRIPE_KEY) + assert masked.startswith("sk_l") + assert masked.endswith("LzXc") + assert "51QwErTyUiOpAsDfGhJk" not in masked + + def test_short_values_are_masked_completely(self): + # A revealed head and tail would be most of a short credential, so + # nothing is kept. + assert mask_value("SuperSecret123!") == "*" * len("SuperSecret123!") + assert mask_value("hunter2") == "*******" + + def test_masked_length_matches_the_original(self): + for value in ("a", "short", STRIPE_KEY): + assert len(mask_value(value)) == len(value) + + def test_empty_and_non_string_values_are_handled(self): + assert mask_value("") == "" + assert mask_value(None) == "" + assert mask_value(12345678901234567890) == "1234************7890" + + +class TestScrubTokens: + @pytest.mark.parametrize("token", SAMPLE_TOKENS) + def test_known_credential_formats_never_survive(self, token): + text = f"value = connect({token})" + assert token not in scrub_tokens(text) + + def test_pem_private_key_bodies_are_replaced(self): + body = _sample( + "MIIEowIBAAKCAQEAwJz9Fq3n0pQ7bTvXyZ1aB2cD3eF4gH5iJ6kL7mN8oP9qR0sT\n", + "uVwXyZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUV", + ) + marker = _sample("-----BEGIN RSA PRIVATE", " KEY-----") + pem = f"{marker}\n{body}\n{marker.replace('BEGIN', 'END')}" + scrubbed = scrub_tokens(pem) + assert body.split("\n")[0] not in scrubbed + assert scrubbed.startswith(marker) + assert scrubbed.endswith(marker.replace("BEGIN", "END")) + + def test_jwt_payloads_are_masked(self): + # The payload segment carries the claims, so it is the part that matters. + payload = JWT.split(".")[1] + assert payload not in scrub_tokens(f"const t = '{JWT}';") + + def test_credentials_in_a_url_authority_are_masked(self): + password = _sample("Tr0ub4dor", "&3xyz") + scrubbed = scrub_tokens(f"postgres://admin:{password}@db.internal:5432/app") + assert password not in scrubbed + # The host stays readable so the finding still points somewhere. + assert "db.internal:5432/app" in scrubbed + + def test_url_password_is_masked_when_it_matches_the_username(self): + scrubbed = scrub_tokens("postgres://admin:admin@db.internal/app") + assert scrubbed.startswith("postgres://admin:") + assert ":admin@" not in scrubbed + + def test_ordinary_code_is_left_alone(self): + for snippet in ( + "eval(rawConfigStr)", + "cursor.execute(f\"SELECT * FROM users WHERE id = {user_id}\")", + "app.use(express.static('public'))", + "if len(password) < 8:", + ): + assert scrub_tokens(snippet) == snippet + + +class TestRedactLiterals: + def test_the_assignment_target_survives_but_the_value_does_not(self): + redacted = redact_literals(f'STRIPE_SECRET_KEY = "{STRIPE_KEY}"') + assert redacted.startswith('STRIPE_SECRET_KEY = "') + assert "51QwErTyUiOpAsDfGhJk" not in redacted + + def test_every_quote_style_is_covered(self): + assert "SuperSecret123!" not in redact_literals( + "const DB_PASSWORD = 'SuperSecret123!';" + ) + assert "SuperSecret123!" not in redact_literals( + "const DB_PASSWORD = `SuperSecret123!`;" + ) + assert "SuperSecret123!" not in redact_literals( + 'DB_PASSWORD = "SuperSecret123!"' + ) + + def test_escaped_quotes_inside_a_literal_do_not_end_it(self): + redacted = redact_literals(r'key = "abc\"def SuperSecret123!"') + assert "SuperSecret123!" not in redacted + + def test_multi_character_operators_keep_their_syntax(self): + """A quoted value must reach the literal pass whatever precedes it. + + Matching only the first character of ``:=`` or ``==`` leaves the rest of + the operator heading the value, which stops looking quoted and sends the + line to the unquoted branch -- which stars out the operator and quotes + the literal pass exists to keep. ``go-hardcoded-credentials`` matches + ``$VAR := "..."``, so this is a shape the bundled rules produce. + """ + for line, prefix in ( + ('apiKey := "SuperSecret123!"', 'apiKey := "'), + ('if password == "SuperSecret123!" {', 'if password == "'), + ('if password != "SuperSecret123!" {', 'if password != "'), + ): + redacted = redact_literals(line) + assert "SuperSecret123!" not in redacted + assert redacted.startswith(prefix), redacted + + def test_a_scope_operator_is_not_read_as_an_assignment(self): + # ``::`` is not an assignment, so the line keeps its shape and the + # literal pass handles the value. + redacted = redact_literals('let cfg = Config::new("SuperSecret123!");') + assert "SuperSecret123!" not in redacted + + def test_unquoted_assignments_fall_back_to_masking_the_value(self): + redacted = redact_literals("password: SuperSecret123!") + assert "SuperSecret123!" not in redacted + assert redacted.startswith("password: ") + + def test_quoted_text_does_not_skip_an_unquoted_value(self): + redacted = redact_literals('password: hunter2 # "temporary"') + assert "hunter2" not in redacted + assert redacted.startswith("password: ") + + def test_each_line_uses_the_appropriate_redaction(self): + redacted = redact_literals( + 'credentials:\n user: "admin"\n password: hunter2' + ) + assert "admin" not in redacted + assert "hunter2" not in redacted + + def test_empty_literals_are_left_as_they_are(self): + assert redact_literals('password = ""') == 'password = ""' + + +class TestCredentialRuleSelection: + @pytest.mark.parametrize( + "rule_id", + [ + "python-hardcoded-secret", + "js-hardcoded-secret", + "java-hardcoded-credentials", + "go-hardcoded-credentials", + "swift-hardcoded-secrets", + "python-hardcoded-password-default", + "js-default-credentials", + "js-weak-jwt-secret", + ], + ) + def test_hardcoded_credential_rules_are_selected(self, rule_id): + assert is_credential_finding(rule_id, {}) + + @pytest.mark.parametrize( + "rule_id", + [ + # The match is a literal address, and masking it would remove the + # reason the finding was raised. + "python-hardcoded-ip", + "java-hardcoded-ip", + # These match a length comparison, not a credential. + "python-weak-password-validation", + "js-weak-password-validation", + "python-sql-injection-format", + "js-eval-usage", + ], + ) + def test_rules_whose_match_is_not_a_credential_are_not_selected(self, rule_id): + assert not is_credential_finding(rule_id, {}) + + def test_rule_metadata_overrides_the_name(self): + assert is_credential_finding("custom-vault-lookup", {"redact": True}) + assert is_credential_finding("custom-vault-lookup", {"redact": "yes"}) + assert not is_credential_finding("python-hardcoded-secret", {"redact": False}) + + def test_every_bundled_hardcoded_credential_rule_is_covered(self): + """The bundled rules are the contract, so check them rather than a list. + + A new language file lands with the same ``*-hardcoded-secrets`` naming + as the existing fifteen; this fails if one arrives that the selector + does not recognize. + """ + uncovered = [] + for rule_file in sorted(glob.glob(str(RULES_DIR / "*.yml"))): + for rule in (yaml.safe_load(Path(rule_file).read_text()) or {}).get("rules", []): + rule_id = rule.get("id", "") + if not re.search(r"hardcoded-(secret|credential|password|key|token)", rule_id): + continue + if not is_credential_finding(rule_id, rule.get("metadata") or {}): + uncovered.append(rule_id) + assert uncovered == [] + + +class TestRedactSnippet: + def test_a_credential_finding_masks_the_literal(self): + redacted = redact_snippet( + 'DB_PASSWORD = "SuperSecret123!"', credential_finding=True + ) + assert "SuperSecret123!" not in redacted + assert "DB_PASSWORD" in redacted + + def test_a_non_credential_finding_keeps_its_code_readable(self): + snippet = "cursor.execute('SELECT * FROM users WHERE id = ' + user_id)" + assert redact_snippet(snippet, credential_finding=False) == snippet + + def test_a_non_credential_finding_still_loses_a_recognizable_token(self): + # The rule is about logging, but the line happens to carry a real key. + snippet = f"console.log('key', '{AWS_KEY_ID}')" + redacted = redact_snippet(snippet, credential_finding=False) + assert AWS_KEY_ID not in redacted + assert "console.log" in redacted + + +class TestRedactMessage: + def test_interpolated_metavariables_are_masked_for_credential_findings(self): + secret = "SuperSecret123!" + redacted = redact_message( + f"Hardcoded credential {secret} assigned to database_password", + { + "$VALUE": {"abstract_content": secret}, + "$VAR": {"abstract_content": "database_password"}, + }, + credential_finding=True, + ) + assert secret not in redacted + assert "database_password" not in redacted + assert redacted.startswith("Hardcoded credential ") + + def test_known_tokens_are_scrubbed_from_every_message(self): + assert AWS_KEY_ID not in redact_message(f"Logged value: {AWS_KEY_ID}") + + +class TestRedactMessage: + def test_a_short_bound_value_does_not_mangle_the_rest_of_the_message(self): + """The replace is by value, so a short one is also an ordinary substring. + + Masking every occurrence would rewrite words that merely contain it. + The credential still has to be masked, so the replace is anchored rather + than skipped. + """ + redacted = redact_message( + "secret a is bad", {"$X": {"abstract_content": "a"}}, credential_finding=True + ) + assert redacted == "secret * is bad" + + def test_a_short_bound_value_is_still_masked(self): + for message, expected in ( + ("password admin is a bad default", "password ***** is a bad default"), + ("key is at the end: admin", "key is at the end: *****"), + ): + assert ( + redact_message( + message, {"$P": {"abstract_content": "admin"}}, credential_finding=True + ) + == expected + ) + + def test_a_long_bound_value_is_masked_wherever_it_appears(self): + redacted = redact_message( + 'Hardcoded secret in DB_PASSWORD = "SuperSecret123!"', + { + "$VAR": {"abstract_content": "DB_PASSWORD"}, + "$V": {"abstract_content": "SuperSecret123!"}, + }, + credential_finding=True, + ) + assert "SuperSecret123!" not in redacted + assert "DB_PASSWORD" not in redacted + + def test_a_non_credential_finding_keeps_its_message(self): + message = "Use of eval() on untrusted input" + assert redact_message(message, {"$X": {"abstract_content": "eval"}}) == message + + +class TestRedactDataflowTrace: + def test_a_credential_finding_masks_literals_in_its_trace(self): + """A trace step is a source line, so it gets the snippet's treatment. + + ``scrub_tokens`` alone would keep a generic password, which the + vendor-format patterns do not recognize. + """ + trace = { + "source": {"content": 'password = "SuperSecret123!"', "file": "a.py", "line": 1}, + "intermediates": [ + {"content": 'tmp = "SuperSecret123!"', "file": "a.py", "line": 2} + ], + "sink": {"content": 'connect(password="SuperSecret123!")', "file": "a.py", "line": 3}, + } + serialized = json.dumps(redact_dataflow_trace(trace, credential_finding=True)) + assert "SuperSecret123!" not in serialized + assert '"line": 3' in serialized + + def test_a_non_credential_finding_keeps_its_trace_readable(self): + trace = { + "source": {"content": "user_id = request.args.get('id')", "file": "a.py", "line": 1}, + "sink": {"content": "cursor.execute(query)", "file": "a.py", "line": 2}, + } + redacted = redact_dataflow_trace(trace) + assert redacted["source"]["content"] == "user_id = request.args.get('id')" + assert redacted["sink"]["content"] == "cursor.execute(query)" + + def test_trace_steps_are_scrubbed(self): + trace = { + "source": {"content": f"key = '{AWS_KEY_ID}'", "file": "a.py", "line": 1}, + "intermediates": [ + {"content": f"tmp = '{GITHUB_TOKEN}'", "file": "a.py", "line": 2} + ], + "sink": {"content": "requests.get(url, headers={'k': key})", "file": "a.py", "line": 3}, + } + serialized = json.dumps(redact_dataflow_trace(trace)) + assert AWS_KEY_ID not in serialized + assert GITHUB_TOKEN not in serialized + # Locations survive, which is what makes the trace useful. + assert '"line": 3' in serialized + + def test_a_trace_of_an_unexpected_shape_is_returned_unchanged(self): + assert redact_dataflow_trace(None) is None + assert redact_dataflow_trace("not a trace") == "not a trace" + + +class TestConnectorOutput: + """The masking has to survive the trip through the connectors.""" + + def test_opengrep_alerts_carry_no_credential(self, tmp_path): + from socket_basics.core.connector.opengrep import OpenGrepScanner + + raw = { + "results": [ + { + "check_id": "socket_basics.rules.python-hardcoded-secret", + "path": str(tmp_path / "config.py"), + "start": {"line": 3}, + "end": {"line": 3}, + "extra": { + "severity": "ERROR", + "message": f"Hardcoded credential detected: {STRIPE_KEY}", + "lines": f'STRIPE_SECRET_KEY = "{STRIPE_KEY}"', + "metavars": { + "$VALUE": {"abstract_content": STRIPE_KEY}, + }, + "metadata": {"cwe": "CWE-798", "confidence": "medium"}, + }, + } + ] + } + + scanner = OpenGrepScanner.__new__(OpenGrepScanner) + scanner.config = _StubConfig(tmp_path) + scanner.allowed_severities = {"critical", "high", "medium", "low"} + components = scanner._convert_to_socket_facts(raw) + + serialized = json.dumps(components) + assert STRIPE_KEY not in serialized + assert "python-hardcoded-secret" in serialized + + def test_trufflehog_alerts_carry_no_credential(self, tmp_path): + from socket_basics.core.connector.trufflehog import TruffleHogScanner + + # TruffleHog reports the match verbatim in Raw and leaves Redacted + # empty for most detectors, so the connector cannot pass either through. + finding = { + "DetectorName": "Stripe", + "Verified": False, + "Raw": STRIPE_KEY, + "Redacted": "", + "SourceMetadata": { + "Data": {"Filesystem": {"file": str(tmp_path / "config.py"), "line": 3}} + }, + } + + scanner = TruffleHogScanner.__new__(TruffleHogScanner) + scanner.config = _StubConfig(tmp_path) + alert = scanner._create_alert(finding) + + serialized = json.dumps(alert) + assert STRIPE_KEY not in serialized + assert alert["props"]["lineNumber"] == 3 + + def test_a_short_secret_is_not_partly_revealed(self, tmp_path): + from socket_basics.core.connector.trufflehog import TruffleHogScanner + + scanner = TruffleHogScanner.__new__(TruffleHogScanner) + scanner.config = _StubConfig(tmp_path) + alert = scanner._create_alert( + { + "DetectorName": "Generic", + "Verified": False, + "Raw": "SuperSecret123!", + "SourceMetadata": { + "Data": {"Filesystem": {"file": str(tmp_path / "a.py"), "line": 1}} + }, + } + ) + assert alert["props"]["redactedValue"] == "*" * len("SuperSecret123!") + + +class _StubConfig: + """The slice of Config the alert builders touch.""" + + def __init__(self, workspace): + self.workspace = workspace + self.output_dir = workspace + self._config = {} + + def get(self, key, default=None): + return self._config.get(key, default) + + def get_action_for_severity(self, severity): + return {"critical": "error", "high": "warn", "medium": "warn", "low": "ignore"}[severity] diff --git a/tests/test_trufflehog_excludes.py b/tests/test_trufflehog_excludes.py index e0c54f2..1dc79b0 100644 --- a/tests/test_trufflehog_excludes.py +++ b/tests/test_trufflehog_excludes.py @@ -176,7 +176,11 @@ def fake_run(command, **kwargs): command = captured["command"] assert command.count("--exclude-paths") == 1 assert captured["exists_during_run"] is True - assert len(captured["contents"]) == 3 + # Three configured directories, plus the facts file this run writes and the + # temporary name it is staged under. + facts_patterns = [p for p in captured["contents"] if "facts" in p] + assert len(facts_patterns) == 2 + assert len(captured["contents"]) == 5 assert not Path(command[command.index("--exclude-paths") + 1]).exists() @@ -208,6 +212,17 @@ def fake_run(command, **kwargs): assert command[-1] == str(tmp_path) +def test_output_file_patterns_are_absolute_for_relative_output_dir(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + scanner = _scanner(Path("."), "") + scanner.config.output_dir = Path(".") + + patterns = scanner._output_file_patterns() + + assert scanner._path_matches_patterns(str(tmp_path / ".socket.facts.json"), patterns) + assert scanner._path_matches_patterns(str(tmp_path / ".socket.facts.json.tmp"), patterns) + + def test_process_results_strips_absolute_workspace_from_output_and_id(tmp_path): scanner = _scanner(tmp_path, "") scanner.generate_notifications = lambda components: {}