diff --git a/packages/tangle-cli/src/tangle_cli/component_inspector.py b/packages/tangle-cli/src/tangle_cli/component_inspector.py index 92f16c5..8adc722 100644 --- a/packages/tangle-cli/src/tangle_cli/component_inspector.py +++ b/packages/tangle-cli/src/tangle_cli/component_inspector.py @@ -11,6 +11,7 @@ from tangle_cli.handler import TangleCliHandler from tangle_cli.models import ComponentInfo, ComponentSpec +from tangle_cli.utils import _normalize_git_url if TYPE_CHECKING: from tangle_cli.client import TangleApiClient @@ -416,7 +417,8 @@ def transparency_check(spec: ComponentSpec) -> tuple[bool, str]: if ann.get("git_remote_url") and ( ann.get("component_yaml_path") or ann.get("git_relative_dir") ): - return True, f"git source metadata links to {ann['git_remote_url']}" + safe_url = _normalize_git_url(ann["git_remote_url"]) + return True, f"git source metadata links to {safe_url}" impl = spec.implementation or {} container = impl.get("container", {}) @@ -447,6 +449,10 @@ def _resolve_git_source(spec: ComponentSpec) -> dict[str, Any] | None: if not git_url: return None + # Server-supplied annotations may still carry credentials in the remote + # URL; strip them before building any browsable/emitted links. + git_url = _normalize_git_url(git_url) + sha = annotations.get("git_remote_sha", "") branch = annotations.get("git_remote_branch", "main") component_yaml = annotations.get("component_yaml_path") diff --git a/packages/tangle-cli/src/tangle_cli/component_publisher.py b/packages/tangle-cli/src/tangle_cli/component_publisher.py index 5db339d..be7923b 100644 --- a/packages/tangle-cli/src/tangle_cli/component_publisher.py +++ b/packages/tangle-cli/src/tangle_cli/component_publisher.py @@ -164,7 +164,15 @@ def __init__( self._git_root = str(git_root or git_info.get("_git_root") or "") or None self.git_remote_sha = git_remote_sha or git_info.get("git_remote_sha") self.git_remote_branch = git_remote_branch or git_info.get("git_remote_branch") - self.git_remote_url = git_remote_url or git_info.get("git_remote_url") + resolved_git_remote_url = git_remote_url or git_info.get("git_remote_url") + # Sanitize here as well as in get_git_info: a caller-supplied + # --git-remote-url bypasses get_git_info's normalization, so this is the + # single choke point that guarantees no credentials are persisted. + self.git_remote_url = ( + utils._normalize_git_url(resolved_git_remote_url) + if resolved_git_remote_url + else resolved_git_remote_url + ) self.git_repo = git_repo def _component_spec_model(self) -> type[Any]: diff --git a/packages/tangle-cli/src/tangle_cli/utils.py b/packages/tangle-cli/src/tangle_cli/utils.py index 0449d45..1261c8d 100644 --- a/packages/tangle-cli/src/tangle_cli/utils.py +++ b/packages/tangle-cli/src/tangle_cli/utils.py @@ -870,31 +870,143 @@ def normalize_annotation_paths( _CI_REPO_URL_VARS: tuple[str, ...] = ("BUILDKITE_REPO", "GITHUB_SERVER_URL", "CI_REPOSITORY_URL") -def _normalize_git_url(url: str) -> str: - """Normalize a git remote URL to a browsable HTTPS URL. +# Query-string parameter names that carry authentication material. Exact +# matches for short/ambiguous keys that must not be caught by the substring +# rules below (e.g. ``sig``, ``key``, ``auth``). +_SENSITIVE_QUERY_KEYS: frozenset[str] = frozenset({ + "access_token", "personal_access_token", "private_token", "token", + "api_key", "apikey", "key", "auth", "authorization", + "password", "passwd", "pwd", "secret", + "sig", "signature", "x-access-token", +}) + +# Fail-closed substrings: any query key containing one of these (case-folded) +# is treated as credential-bearing even when it is not an exact known key. This +# catches provider-specific and future keys such as ``oauth_token``, +# ``X-Amz-Signature``, ``X-Amz-Credential`` or ``X-Amz-Security-Token`` without +# over-matching benign params like ``ref``/``path`` — which is why bare +# ``key``/``sig``/``auth`` stay exact-match only above. +_SENSITIVE_QUERY_SUBSTRINGS: tuple[str, ...] = ( + "token", "secret", "password", "passwd", "credential", + "signature", "apikey", "api_key", "oauth", "x-amz-", "x-access", +) + + +def _is_sensitive_query_key(key: str) -> bool: + """Return whether a URL query parameter name carries authentication material.""" + folded = key.lower() + if folded in _SENSITIVE_QUERY_KEYS: + return True + return any(token in folded for token in _SENSITIVE_QUERY_SUBSTRINGS) - Handles common formats: - - ``git@github.com:Org/repo.git`` -> ``https://github.com/Org/repo`` - - ``ssh://git@github.com/Org/repo.git`` -> ``https://github.com/Org/repo`` - - ``https://github.com/Org/repo.git`` -> ``https://github.com/Org/repo`` - - ``https://github.com/Org/repo`` -> unchanged - The ``.git`` suffix is stripped so the result can be used directly to - build ``/blob/{ref}/{path}`` links without an extra ``.removesuffix``. +def _redact_sensitive_query(query: str) -> str: + """Drop credential-bearing parameters from a URL query string. + + Uses a fail-closed predicate (:func:`_is_sensitive_query_key`) so unknown + credential-shaped keys are dropped rather than allowed through. A query + with no sensitive keys is returned byte-for-byte unchanged so that ordinary + URLs are not silently re-encoded. """ - import re + if not query: + return query + + from urllib.parse import parse_qsl, urlencode + + pairs = parse_qsl(query, keep_blank_values=True) + if not any(_is_sensitive_query_key(key) for key, _ in pairs): + return query + kept = [(key, value) for key, value in pairs if not _is_sensitive_query_key(key)] + return urlencode(kept) + + +# Placeholder emitted when a URL-like remote carries credentials but cannot be +# parsed into a clean host (malformed authority, missing host with userinfo, +# malformed IPv6). Returning this rather than the raw input keeps credential +# material out of persisted annotations, CLI output, logs, and browse links, and +# stops the parser from raising into callers. +_REDACTED_GIT_URL: str = "[redacted-invalid-git-url]" - # SCP-style: git@host:path - m = re.match(r"^git@([^:]+):(.+)$", url) - if m: - url = f"https://{m.group(1)}/{m.group(2)}" - else: - # ssh://git@host/path - m = re.match(r"^ssh://(?:[^@]+@)?([^/]+)/(.+)$", url) - if m: - url = f"https://{m.group(1)}/{m.group(2)}" - return url.removesuffix(".git") +def _normalize_git_url(url: str) -> str: + """Normalize a git remote URL to a browsable, credential-free HTTPS URL. + + Converts SSH/SCP forms to HTTPS and strips the ``.git`` suffix so the + result can build ``/blob/{ref}/{path}`` links directly: + + - ``git@github.com:Org/repo.git`` -> ``https://github.com/Org/repo`` + - ``ssh://git@github.com/Org/repo.git`` -> ``https://github.com/Org/repo`` + - ``https://github.com/Org/repo.git`` -> ``https://github.com/Org/repo`` + - ``https://github.com/Org/repo`` -> unchanged + + Any embedded credentials are removed: ``user:password@`` / ``token@`` + userinfo is stripped from URL-form and scheme-relative remotes and dropped + entirely from SCP-style remotes, and sensitive query parameters are redacted. + This guarantees secrets never reach persisted annotations, CLI output, logs, + or error messages. Host, port, path, and fragment are preserved. + + Parsing fails closed: a URL-like input whose credential-bearing authority + cannot be parsed into a clean host (missing host with userinfo, malformed + IPv6, and similar) yields ``_REDACTED_GIT_URL`` rather than leaking the raw + ``user:secret@`` text or raising; a malformed textual port is dropped while + the credential-free host is kept. Local filesystem paths and hostless + schemes (e.g. ``file:///path``) are returned unchanged (aside from ``.git`` + stripping). The function is idempotent. + """ + from urllib.parse import urlsplit, urlunsplit + + if not url: + return url + + stripped = url.strip() + + # SCP-style syntax: ``[user@]host:path`` (no scheme). We drop any userinfo + # since it is authentication material, and rewrite to https so the result + # is browsable. Guard against Windows drive paths (``C:\...``) and against + # anything that already carries an explicit scheme. + if "://" not in stripped and not re.match(r"^[A-Za-z]:[\\/]", stripped): + scp = re.match(r"^(?:[^@/]+@)?([^/:]+):(?!//)(.+)$", stripped) + if scp: + stripped = f"https://{scp.group(1)}/{scp.group(2)}" + + # URL-like inputs (explicit scheme or scheme-relative ``//authority``) must + # fail closed; a bare local path never leaks userinfo, so it is exempt. + url_like = "://" in stripped or stripped.startswith("//") + + try: + parts = urlsplit(stripped) + host = parts.hostname + try: + port = parts.port + except ValueError: + # Malformed textual port (e.g. ``host:notaport``): drop the port but + # keep the credential-free host so the link stays browsable. + port = None + except ValueError: + # Malformed authority (e.g. unterminated IPv6 ``[::1``). + return _REDACTED_GIT_URL if url_like else stripped.removesuffix(".git") + + if parts.scheme or parts.netloc: + if host is None: + # URL-form/scheme-relative with no parseable host. If the authority + # carried userinfo, returning the raw text would leak it — fail + # closed. Otherwise it is a legitimately hostless scheme (file://). + if "@" in parts.netloc: + return _REDACTED_GIT_URL + return stripped.removesuffix(".git") + # Re-bracket IPv6 literals, which ``hostname`` returns without brackets. + if ":" in host and not host.startswith("["): + host = f"[{host}]" + netloc = host if port is None else f"{host}:{port}" + out_scheme = "https" if parts.scheme.lower() == "ssh" else parts.scheme.lower() + query = _redact_sensitive_query(parts.query) + # Strip ``.git`` from the path itself so it is removed even when a query + # or fragment follows it. + path = parts.path.removesuffix(".git") + return urlunsplit((out_scheme, netloc, path, query, parts.fragment)) + + # No scheme and no authority: a bare local path. + return stripped.removesuffix(".git") def _fill_from_ci_env(info: dict[str, str]) -> None: diff --git a/tests/test_component_inspector.py b/tests/test_component_inspector.py index 7d5c068..f3dc465 100644 --- a/tests/test_component_inspector.py +++ b/tests/test_component_inspector.py @@ -97,6 +97,27 @@ def test_unknown_container_is_opaque(self): assert transparent is False assert "no inline source" in reason + def test_git_source_reason_does_not_leak_credentials(self): + spec = ComponentSpec.from_dict({ + "spec": { + "name": "demo", + "implementation": {"container": {"image": "registry.example.com/private/demo:latest"}}, + "metadata": { + "annotations": { + "git_remote_url": "https://user:s3cr3tTOKEN@github.com/Org/repo.git", + "component_yaml_path": "comp.yaml", + }, + }, + }, + }) + + transparent, reason = ComponentInspector.transparency_check(spec) + + assert transparent is True + assert "s3cr3tTOKEN" not in reason + assert "@github.com" not in reason + assert "https://github.com/Org/repo" in reason + class TestComponentLibrary: def test_standard_library_does_not_fetch_cross_origin_component_urls(self): diff --git a/tests/test_component_publisher.py b/tests/test_component_publisher.py index a6ac801..0f56dc9 100644 --- a/tests/test_component_publisher.py +++ b/tests/test_component_publisher.py @@ -396,6 +396,20 @@ def test_publish_components_passes_structured_context_to_context_aware_hooks(tmp assert kwargs_hook.contexts == [after_context] +def test_publisher_strips_credentials_from_supplied_git_remote_url(tmp_path: Path) -> None: + # A caller-supplied --git-remote-url bypasses get_git_info's normalization, + # so the publisher itself must sanitize it before it reaches annotations. + publisher = ComponentPublisher( + dry_run=True, + git_remote_url="https://gitlab-ci-token:glcbt-SECRETVALUE@gitlab.com/Org/repo.git", + git_root=tmp_path, + ) + + assert publisher.git_remote_url == "https://gitlab.com/Org/repo" + assert "glcbt-SECRETVALUE" not in (publisher.git_remote_url or "") + assert "@" not in (publisher.git_remote_url or "") + + def test_publish_components_batches_configs_and_runs_hooks(tmp_path: Path) -> None: first = write_component(tmp_path / "one.yaml", name="one", version="1.0") second = write_component(tmp_path / "two.yaml", name="two", version="2.0") diff --git a/tests/test_utils.py b/tests/test_utils.py index 24f6a4c..035c7d0 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -11,6 +11,7 @@ import pytest from tangle_cli.utils import ( + _REDACTED_GIT_URL, UnsetVarError, _normalize_git_url, apply_defaults, @@ -177,3 +178,187 @@ class TestNormalizeGitUrl: ]) def test_normalization(self, input_url, expected): assert _normalize_git_url(input_url) == expected + + @pytest.mark.parametrize("input_url,expected", [ + # user:password userinfo is stripped from http(s) URLs + ( + "https://user:s3cr3t@github.com/Org/repo.git", + "https://github.com/Org/repo", + ), + # token-style single-field userinfo (personal access token) + ( + "https://ghp_ABC123token@github.com/Org/repo.git", + "https://github.com/Org/repo", + ), + # username-only userinfo + ( + "https://alice@example.com/Org/repo.git", + "https://example.com/Org/repo", + ), + # password-only / empty username + ( + "https://:onlypassword@example.com/Org/repo", + "https://example.com/Org/repo", + ), + # percent-encoded userinfo (e.g. an email-style username and @ in secret) + ( + "https://user%40corp.com:p%40ss%2Fword@gitlab.com/Org/repo.git", + "https://gitlab.com/Org/repo", + ), + # plain http is preserved as http (scheme not silently upgraded) + ( + "http://user:pw@internal.example/Org/repo.git", + "http://internal.example/Org/repo", + ), + # host + port is preserved while credentials are removed + ( + "https://user:pw@example.com:8443/Org/repo.git", + "https://example.com:8443/Org/repo", + ), + # GitLab CI token URL (a very common real-world leak vector) + ( + "https://gitlab-ci-token:glcbt-xxxxxxxx@gitlab.com/Org/repo.git", + "https://gitlab.com/Org/repo", + ), + # ssh:// with userinfo -> https, credentials dropped + ( + "ssh://git@github.com/Org/repo.git", + "https://github.com/Org/repo", + ), + # ssh:// with an explicit port keeps the port + ( + "ssh://git@github.com:2222/Org/repo.git", + "https://github.com:2222/Org/repo", + ), + # scp-style with a username other than git + ( + "deploy@example.com:Org/repo.git", + "https://example.com/Org/repo", + ), + # IPv6 literal host with credentials and port + ( + "https://user:pw@[2001:db8::1]:8443/Org/repo.git", + "https://[2001:db8::1]:8443/Org/repo", + ), + # fragment is preserved + ( + "https://user:pw@github.com/Org/repo.git#readme", + "https://github.com/Org/repo#readme", + ), + ]) + def test_credentials_are_stripped(self, input_url, expected): + assert _normalize_git_url(input_url) == expected + + @pytest.mark.parametrize("secret,input_url", [ + ("s3cr3t", "https://user:s3cr3t@github.com/Org/repo.git"), + ("ghp_ABC123token", "https://ghp_ABC123token@github.com/Org/repo.git"), + ("glcbt-xxxxxxxx", "https://gitlab-ci-token:glcbt-xxxxxxxx@gitlab.com/Org/repo.git"), + ("p%40ss%2Fword", "https://u:p%40ss%2Fword@gitlab.com/Org/repo.git"), + ("onlypassword", "https://:onlypassword@example.com/Org/repo"), + ]) + def test_no_secret_material_survives(self, secret, input_url): + result = _normalize_git_url(input_url) + assert secret not in result + assert "@" not in result + + @pytest.mark.parametrize("input_url,expected", [ + # sensitive query parameters are redacted + ( + "https://github.com/Org/repo?access_token=abc123", + "https://github.com/Org/repo", + ), + ( + "https://example.com/Org/repo.git?private_token=tok&ref=main", + "https://example.com/Org/repo?ref=main", + ), + ( + "https://example.com/Org/repo?password=hunter2&x=1", + "https://example.com/Org/repo?x=1", + ), + ]) + def test_sensitive_query_params_redacted(self, input_url, expected): + assert _normalize_git_url(input_url) == expected + + @pytest.mark.parametrize("input_url", [ + "https://github.com/Org/repo", + "https://github.com/Org/repo?ref=main&path=a/b", + "/local/path/to/repo", + "./relative/repo", + "file:///home/user/repo", + "git@github.com:Org/repo.git", + ]) + def test_credential_free_urls_are_preserved(self, input_url): + # Non-sensitive query strings and local paths must not be corrupted. + result = _normalize_git_url(input_url) + assert _normalize_git_url(result) == result # idempotent + + def test_local_paths_not_corrupted(self): + assert _normalize_git_url("/abs/path/repo") == "/abs/path/repo" + assert _normalize_git_url("./rel/repo") == "./rel/repo" + assert _normalize_git_url("file:///home/user/repo.git") == "file:///home/user/repo" + + def test_windows_path_not_treated_as_scp(self): + assert _normalize_git_url(r"C:\Users\me\repo") == r"C:\Users\me\repo" + + def test_empty_and_whitespace(self): + assert _normalize_git_url("") == "" + assert _normalize_git_url(" https://user:pw@github.com/Org/repo.git ") == ( + "https://github.com/Org/repo" + ) + + def test_idempotent(self): + once = _normalize_git_url("https://user:token@github.com/Org/repo.git") + assert _normalize_git_url(once) == once + + @pytest.mark.parametrize("input_url,expected", [ + # oauth_token is not an exact known key but is credential-shaped + ( + "https://github.com/Org/repo?oauth_token=SECRETVAL", + "https://github.com/Org/repo", + ), + # AWS SigV4 presigned-URL params (mixed case) are dropped fail-closed + ( + "https://host/Org/repo?X-Amz-Signature=SECRETSIG&X-Amz-Credential=AKIA/x&ref=main", + "https://host/Org/repo?ref=main", + ), + ( + "https://host/Org/repo?X-Amz-Security-Token=SECRETTOK&path=a/b", + "https://host/Org/repo?path=a%2Fb", + ), + ]) + def test_unknown_credential_query_keys_dropped_fail_closed(self, input_url, expected): + result = _normalize_git_url(input_url) + assert result == expected + for secret in ("SECRETVAL", "SECRETSIG", "AKIA", "SECRETTOK"): + assert secret not in result + + def test_missing_host_with_userinfo_fails_closed(self): + # scheme present, userinfo present, but no parseable host: must not leak + result = _normalize_git_url("https://user:secret@/Org/repo.git") + assert result == _REDACTED_GIT_URL + assert "secret" not in result + assert "@" not in result + + def test_scheme_relative_userinfo_is_stripped(self): + # ``//user:secret@host/path`` previously fell through with creds intact + result = _normalize_git_url("//user:secret@host/Org/repo.git") + assert result == "//host/Org/repo" + assert "secret" not in result + assert "@" not in result + + def test_invalid_textual_port_does_not_raise(self): + # ``.port`` raises ValueError when read; we drop the bad port, keep host + result = _normalize_git_url("https://user:secret@host:notaport/Org/repo.git") + assert result == "https://host/Org/repo" + assert "secret" not in result + assert "@" not in result + + def test_malformed_ipv6_fails_closed(self): + # urlsplit itself raises on an unterminated IPv6 authority + result = _normalize_git_url("https://user:secret@[::1/Org/repo.git") + assert result == _REDACTED_GIT_URL + assert "secret" not in result + + def test_hostless_file_url_preserved(self): + # a legitimately hostless scheme carries no userinfo and must survive + assert _normalize_git_url("file:///home/user/repo.git") == "file:///home/user/repo"