fix(deps): honor SSH user from dependency URL (closes #1383)

src/apm_cli/deps/host_backends.py

@@ -192,6 +192,7 @@ def build_clone_ssh_url(self, dep_ref: DependencyReference) -> str:
             self._url_host(dep_ref),
             dep_ref.repo_url,
             port=getattr(dep_ref, "port", None),
+            user=getattr(dep_ref, "ssh_user", None) or "git",
         )
 
     def build_clone_http_url(self, dep_ref: DependencyReference) -> str:
@@ -391,7 +392,12 @@ def build_clone_https_url(
 
     def build_clone_ssh_url(self, dep_ref: DependencyReference) -> str:
         host = getattr(dep_ref, "host", None) or self.host_info.host
-        return build_ssh_url(host, dep_ref.repo_url, port=getattr(dep_ref, "port", None))
+        return build_ssh_url(
+            host,
+            dep_ref.repo_url,
+            port=getattr(dep_ref, "port", None),
+            user=getattr(dep_ref, "ssh_user", None) or "git",
+        )
 
     def build_clone_http_url(self, dep_ref: DependencyReference) -> str:
         port = getattr(dep_ref, "port", None)
@@ -469,7 +475,12 @@ def build_clone_https_url(
 
     def build_clone_ssh_url(self, dep_ref: DependencyReference) -> str:
         host = getattr(dep_ref, "host", None) or self.host_info.host
-        return build_ssh_url(host, dep_ref.repo_url, port=getattr(dep_ref, "port", None))
+        return build_ssh_url(
+            host,
+            dep_ref.repo_url,
+            port=getattr(dep_ref, "port", None),
+            user=getattr(dep_ref, "ssh_user", None) or "git",
+        )
 
     def build_clone_http_url(self, dep_ref: DependencyReference) -> str:
         port = getattr(dep_ref, "port", None)

src/apm_cli/models/dependency/reference.py

@@ -17,6 +17,7 @@
     maybe_raise_bare_fqdn_github_gitlab_conflict,
     parse_artifactory_path,
     unsupported_host_error,
+    validate_ssh_user,
 )
 from ...utils.path_security import (
     PathTraversalError,
@@ -88,6 +89,14 @@ class DependencyReference:
     # SKILL_BUNDLE subset selection (persisted in apm.yml `skills:` field)
     skill_subset: list[str] | None = None  # Sorted skill names, or None = all
 
+    # SSH username for SCP-shorthand or ``ssh://`` dependencies. ``None`` for
+    # non-SSH inputs. Defaults to ``"git"`` whenever an SSH form was parsed
+    # without an explicit user. Carried as auth/transport context, NOT
+    # baked into ``to_canonical()`` / ``get_identity()`` so dependency
+    # identity stays user-agnostic (lockfile pinning + dedup work the same
+    # whether a project uses ``git@`` or an EMU/custom SSH account).
+    ssh_user: str | None = None
+
     # Supported file extensions for virtual packages
     VIRTUAL_FILE_EXTENSIONS = (
         ".prompt.md",
@@ -424,12 +433,27 @@ def _parse_ssh_protocol_url(url: str):
             ssh://git@host/owner/repo.git@alias
 
         Returns:
-            ``(host, port, repo_url, reference, alias)`` or ``None`` if the
-            input is not an ``ssh://`` URL.
+            ``(host, port, repo_url, reference, alias, user)`` or ``None`` if
+            the input is not an ``ssh://`` URL. ``user`` defaults to ``"git"``
+            when no userinfo is present.
         """
         if not url.startswith("ssh://"):
             return None
 
+        # SECURITY: reject percent-encoded userinfo BEFORE urlparse decodes it.
+        # ``urlparse('ssh://%2DoProxyCommand=evil@host/repo').username`` returns
+        # ``-oProxyCommand=evil`` which would smuggle SSH options past the
+        # allowlist in validate_ssh_user. We inspect the raw substring between
+        # ``ssh://`` and the first ``@`` (which terminates the userinfo per
+        # RFC 3986) and reject any ``%`` there. There is no legitimate need for
+        # percent-encoding in a real SSH username.
+        userinfo_match = re.match(r"^ssh://([^@/?#]+)@", url)
+        if userinfo_match and "%" in userinfo_match.group(1):
+            raise ValueError(
+                "Percent-encoded characters are not allowed in SSH userinfo. "
+                "Use the literal username (e.g. 'ssh://myuser@host/...')."
+            )
+
         parsed = urllib.parse.urlparse(url)
         host = parsed.hostname or ""
         port = parsed.port  # int or None
@@ -439,6 +463,12 @@ def _parse_ssh_protocol_url(url: str):
         path = parsed.path.lstrip("/")
         fragment = parsed.fragment
 
+        # Userinfo: validate or default to "git". urlparse exposes ``username``
+        # already percent-decoded; the pre-check above guarantees no decoding
+        # actually happened, so what we see equals what was on the wire.
+        raw_user = parsed.username
+        ssh_user = validate_ssh_user(raw_user) if raw_user else "git"
+
         reference: str | None = None
         alias: str | None = None
 
@@ -464,7 +494,7 @@ def _parse_ssh_protocol_url(url: str):
         # Security: reject traversal sequences in SSH repo paths
         validate_path_segments(repo_url, context="SSH repository path", reject_empty=True)
 
-        return host, port, repo_url, reference, alias
+        return host, port, repo_url, reference, alias, ssh_user
 
     @staticmethod
     def _normalize_parent_repo_decl_path(raw: str) -> str:
@@ -972,7 +1002,8 @@ def _parse_ssh_url(dependency_str: str):
         # Security: reject traversal sequences in SSH repo paths
         validate_path_segments(repo_url, context="SSH repository path", reject_empty=True)
 
-        return host, None, repo_url, reference, alias
+        ssh_user = validate_ssh_user(user)
+        return host, None, repo_url, reference, alias, ssh_user
 
     @classmethod
     def _resolve_virtual_shorthand_repo(cls, repo_url, validated_host, virtual_path=None):
@@ -1435,14 +1466,15 @@ def parse(cls, dependency_str: str) -> "DependencyReference":
         # Phase 2: parse SSH (ssh:// URL first -- it preserves port; then SCP
         # shorthand), otherwise fall back to HTTPS/shorthand parsing.
         explicit_scheme: str | None = None
+        ssh_user: str | None = None
         ssh_proto_result = cls._parse_ssh_protocol_url(dependency_str)
         if ssh_proto_result:
-            host, port, repo_url, reference, alias = ssh_proto_result
+            host, port, repo_url, reference, alias, ssh_user = ssh_proto_result
             explicit_scheme = "ssh"
         else:
             scp_result = cls._parse_ssh_url(dependency_str)
             if scp_result:
-                host, port, repo_url, reference, alias = scp_result
+                host, port, repo_url, reference, alias, ssh_user = scp_result
                 explicit_scheme = "ssh"
             else:
                 host, port, repo_url, reference, alias, is_virtual_package, virtual_path = (
@@ -1484,6 +1516,7 @@ def parse(cls, dependency_str: str) -> "DependencyReference":
             ado_repo=ado_repo,
             artifactory_prefix=artifactory_prefix,
             is_insecure=urllib.parse.urlparse(dependency_str).scheme.lower() == "http",
+            ssh_user=ssh_user,
         )
 
     def to_apm_yml_entry(self):

src/apm_cli/utils/github_host.py

@@ -278,17 +278,67 @@ def build_raw_content_url(owner: str, repo: str, ref: str, file_path: str) -> st
     return f"https://raw.githubusercontent.com/{owner}/{repo}/{encoded_ref}/{file_path}"
 
 
-def build_ssh_url(host: str, repo_ref: str, port: int | None = None) -> str:
+_SSH_USER_RE = re.compile(r"^[a-zA-Z0-9_][a-zA-Z0-9_.+-]*$")
+_SSH_USER_MAX_LEN = 64
+
+
+def validate_ssh_user(user: str) -> str:
+    """Validate an SSH username; return it unchanged or raise ``ValueError``.
+
+    Allowlist policy (deliberately strict):
+
+    - First character must be alphanumeric or underscore. This blocks
+      SSH option injection vectors like ``-oProxyCommand=...`` from ever
+      reaching ``git clone`` argv as a userinfo segment.
+    - Remaining characters are letters, digits, ``.``, ``+``, ``-``, ``_``.
+      This forbids ``/`` (path escape), ``@`` (double-userinfo confusion
+      in ``ssh://user@host``), ``:`` (port confusion), and any whitespace
+      or control character (log/ANSI injection).
+    - Maximum length 64 bytes: long enough for any legitimate username
+      and short enough to bound log size and reject buffer-abuse payloads.
+
+    The shape matches the ``user`` group in ``SCP_LIKE_RE``
+    (``cache/url_normalize.py``) so SCP-shorthand inputs that parsed
+    successfully never fail this validation, while ``ssh://`` URLs (whose
+    userinfo is percent-decoded by ``urllib.parse``) are still gated.
+    """
+    if not user:
+        raise ValueError("SSH user must be a non-empty string")
+    if len(user) > _SSH_USER_MAX_LEN:
+        raise ValueError(f"SSH user is too long ({len(user)} > {_SSH_USER_MAX_LEN} chars)")
+    if not _SSH_USER_RE.match(user):
+        # Do NOT echo the raw user value -- a hostile apm.yml could embed
+        # control characters that survive log emission. Show only the length.
+        raise ValueError(
+            f"Invalid SSH user (length {len(user)}). "
+            "Allowed: alphanumerics, '.', '+', '-', '_'; "
+            "must not start with '-'."
+        )
+    return user
+
+
+def build_ssh_url(
+    host: str,
+    repo_ref: str,
+    port: int | None = None,
+    user: str = "git",
+) -> str:
     """Build an SSH clone URL for the given host and repo_ref (owner/repo).
 
     When ``port`` is set, emit the explicit ``ssh://`` form because SCP
     shorthand (``git@host:path``) cannot carry a port — the ``:`` is the path
     separator. Without a port, keep the compact SCP shorthand (no behavioural
     change for the common case).
+
+    ``user`` defaults to ``"git"`` for backward compatibility with public
+    GitHub / GitLab / Bitbucket which all expect that fixed account name.
+    Non-default usernames (EMU SSH accounts, self-hosted servers with a
+    different bot user) are passed through after ``validate_ssh_user``.
     """
+    safe_user = validate_ssh_user(user)
     if port:
-        return f"ssh://git@{host}:{port}/{repo_ref}.git"
-    return f"git@{host}:{repo_ref}.git"
+        return f"ssh://{safe_user}@{host}:{port}/{repo_ref}.git"
+    return f"{safe_user}@{host}:{repo_ref}.git"
 
 
 def build_https_clone_url(

tests/acceptance/test_logging_acceptance.py

@@ -454,7 +454,7 @@ def test_dry_run_no_file_changes(
 
     def test_status_symbols_are_ascii_brackets(self):
         """All STATUS_SYMBOLS must be ASCII bracket format [x]."""
-        bracket_pattern = {"[*]", "[>]", "[i]", "[!]", "[x]", "[+]", "[#]"}
+        bracket_pattern = {"[*]", "[>]", "[i]", "[!]", "[x]", "[+]", "[#]", "[~]", "[-]", "[=]"}
         for key, sym in STATUS_SYMBOLS.items():
             assert sym in bracket_pattern, (
                 f"STATUS_SYMBOLS['{key}'] = '{sym}' is not a valid bracket symbol"

tests/integration/test_dep_url_parsing_e2e.py

@@ -226,3 +226,105 @@ def test_legacy_git_user_in_apm_yml_still_parses(self, tmp_path):
         dep = deps[0]
         assert dep.host == "github.com"
         assert dep.repo_url == "contoso/my-pkg"
+
+
+class TestSshUserThreadedThroughCloneUrl:
+    """Defends issue #1383 end-to-end.
+
+    Pre-fix: writing ``myuser@github.com:org/repo`` in apm.yml resulted in
+    ``apm install`` cloning ``git@github.com:org/repo.git`` (wrong user). The
+    custom user was parsed by regex but discarded; the host backend always
+    asked ``build_ssh_url`` for the hardcoded ``git`` user.
+
+    Post-fix: the SSH user is captured at parse time, validated against the
+    supply-chain allowlist, and threaded through to the clone URL.
+
+    These tests parse through the real ``APMPackage.from_apm_yml`` entry
+    point AND then invoke the real ``GitHubBackend.build_clone_ssh_url`` so
+    a regression in either layer surfaces here.
+    """
+
+    def test_custom_scp_user_survives_through_clone_url(self, tmp_path):
+        from apm_cli.core.auth import HostInfo
+        from apm_cli.deps.host_backends import GitHubBackend
+
+        apm_yml = _write_minimal_apm_yml(tmp_path, deps=["myuser@github.com:contoso/my-pkg"])
+        pkg = APMPackage.from_apm_yml(apm_yml)
+        deps = pkg.get_apm_dependencies()
+        assert len(deps) == 1
+        dep = deps[0]
+        assert dep.ssh_user == "myuser"
+
+        host_info = HostInfo(
+            host="github.com",
+            kind="github",
+            has_public_repos=True,
+            api_base="https://api.github.com",
+        )
+        backend = GitHubBackend(host_info)
+        clone_url = backend.build_clone_ssh_url(dep)
+        assert clone_url == "myuser@github.com:contoso/my-pkg.git"
+
+    def test_custom_ssh_protocol_user_with_port_survives_through_clone_url(self, tmp_path):
+        from apm_cli.core.auth import HostInfo
+        from apm_cli.deps.host_backends import GenericGitBackend
+
+        apm_yml = _write_minimal_apm_yml(
+            tmp_path, deps=["ssh://buildbot@git.corp.com:7999/team/my-pkg.git"]
+        )
+        pkg = APMPackage.from_apm_yml(apm_yml)
+        deps = pkg.get_apm_dependencies()
+        assert len(deps) == 1
+        dep = deps[0]
+        assert dep.ssh_user == "buildbot"
+        assert dep.port == 7999
+
+        host_info = HostInfo(
+            host="git.corp.com",
+            kind="generic",
+            has_public_repos=False,
+            api_base="https://git.corp.com",
+        )
+        backend = GenericGitBackend(host_info)
+        clone_url = backend.build_clone_ssh_url(dep)
+        assert clone_url == "ssh://buildbot@git.corp.com:7999/team/my-pkg.git"
+
+    def test_legacy_git_user_clone_url_unchanged_for_backward_compat(self, tmp_path):
+        """Regression guard: the default ``git@`` SSH user MUST stay the
+        default for any dep written without an explicit user. This is what
+        every existing apm.yml in the wild relies on."""
+        from apm_cli.core.auth import HostInfo
+        from apm_cli.deps.host_backends import GitHubBackend
+
+        apm_yml = _write_minimal_apm_yml(tmp_path, deps=["git@github.com:contoso/my-pkg"])
+        pkg = APMPackage.from_apm_yml(apm_yml)
+        deps = pkg.get_apm_dependencies()
+        host_info = HostInfo(
+            host="github.com",
+            kind="github",
+            has_public_repos=True,
+            api_base="https://api.github.com",
+        )
+        backend = GitHubBackend(host_info)
+        clone_url = backend.build_clone_ssh_url(deps[0])
+        assert clone_url == "git@github.com:contoso/my-pkg.git"
+
+    def test_option_injection_user_in_apm_yml_is_rejected(self, tmp_path):
+        """A malicious or typo'd apm.yml that starts the SSH user with ``-``
+        (which OpenSSH would interpret as an option flag like
+        ``-oProxyCommand=...``) MUST fail at parse time, not silently flow
+        into ``git clone`` argv."""
+        apm_yml = _write_minimal_apm_yml(tmp_path, deps=["-oProxy@github.com:contoso/my-pkg"])
+        with pytest.raises(ValueError):
+            APMPackage.from_apm_yml(apm_yml).get_apm_dependencies()
+
+    def test_percent_encoded_ssh_user_is_rejected(self, tmp_path):
+        """``urlparse('ssh://%2DoProxyCommand@host/repo').username`` returns
+        ``-oProxyCommand`` — the percent-decode would smuggle option injection
+        past the allowlist. The parser must reject percent-encoded userinfo
+        BEFORE urlparse decodes it."""
+        apm_yml = _write_minimal_apm_yml(
+            tmp_path, deps=["ssh://%2DoProxyCommand@github.com/contoso/my-pkg.git"]
+        )
+        with pytest.raises(ValueError):
+            APMPackage.from_apm_yml(apm_yml).get_apm_dependencies()

tests/unit/test_generic_git_urls.py

@@ -393,6 +393,41 @@ def test_https_clone_url_with_custom_port(self):
         url = build_https_clone_url("bitbucket.domain.ext", "team/repo", port=8443)
         assert url == "https://bitbucket.domain.ext:8443/team/repo"
 
+    def test_ssh_clone_url_with_custom_user(self):
+        """Custom SSH usernames (e.g. EMU accounts) are preserved in the SCP shorthand."""
+        url = build_ssh_url("github.com", "acme/repo", user="myuser")
+        assert url == "myuser@github.com:acme/repo.git"
+
+    def test_ssh_clone_url_with_custom_user_and_port(self):
+        """Custom SSH user + port emits the explicit ssh:// form."""
+        url = build_ssh_url("bitbucket.domain.ext", "team/repo", port=7999, user="myuser")
+        assert url == "ssh://myuser@bitbucket.domain.ext:7999/team/repo.git"
+
+    def test_ssh_clone_url_default_user_unchanged(self):
+        """Omitting the user keeps the historical ``git@`` default."""
+        url = build_ssh_url("github.com", "acme/repo")
+        assert url == "git@github.com:acme/repo.git"
+
+    def test_ssh_clone_url_rejects_option_injection_user(self):
+        """A leading ``-`` would be interpreted as an SSH option flag by OpenSSH; reject it."""
+        import pytest
+
+        with pytest.raises(ValueError, match="Invalid SSH user"):
+            build_ssh_url("github.com", "acme/repo", user="-oProxyCommand=evil")
+
+    def test_ssh_clone_url_rejects_user_with_at_sign(self):
+        """A ``@`` in the user would split the userinfo and shift the host."""
+        import pytest
+
+        with pytest.raises(ValueError, match="Invalid SSH user"):
+            build_ssh_url("github.com", "acme/repo", user="user@other-host")
+
+    def test_ssh_clone_url_rejects_empty_user(self):
+        import pytest
+
+        with pytest.raises(ValueError, match="non-empty"):
+            build_ssh_url("github.com", "acme/repo", user="")
+
     def test_https_clone_url_with_token_and_port(self):
         url = build_https_clone_url("bitbucket.domain.ext", "team/repo", token="pat-xxx", port=8443)
         assert url == "https://x-access-token:pat-xxx@bitbucket.domain.ext:8443/team/repo.git"