Feat/oauth resolution

README.md

@@ -123,6 +123,17 @@ v0.6 and later use the risk-based output and the `2026-07-10` analysis API.
 
 Both versions scan MCP servers, tools, prompts, resources, and skills, and automatically discover supported agent configurations such as Claude Code/Desktop, Cursor, Gemini CLI, and Windsurf.
 
+Or scan exactly one MCP server, skipping every other server and all skills:
+
+```bash
+# one configured server, by name
+uvx snyk-agent-scan@latest scan --server MY_SERVER
+# a remote server by URL, with the transport pinned so nothing is probed
+uvx snyk-agent-scan@latest scan --url https://example.com/mcp --server-type http
+```
+
+See the [CLI reference](docs/cli-reference.md#targeting-a-single-mcp-server-scan-inspect) for the full set of targeting options, including the `npm:` / `pypi:` / `oci:` prefixes for scanning a server straight from a package.
+
 ### Run with a standalone binary
 
 Download the binary for your operating system and architecture from the [latest GitHub Release](https://github.com/snyk/agent-scan/releases/latest). The release page also provides an SBOM (`sbom-<version>.json`), checksum files, and GitHub-generated source code archives. See [Verifying Standalone Binaries](#verifying-standalone-binaries) to verify your download.

docs/cli-reference.md

@@ -84,6 +84,50 @@ snyk-agent-scan evo [CONFIG_FILE ...]
 
 With `--no-skills`, explicit skill paths, skill directories, and automatically discovered skills are all skipped.
 
+### Targeting a single MCP server (`scan`, `inspect`)
+
+| Flag | Type | Default | Description |
+| --- | --- | --- | --- |
+| `--server NAME` | string | none | Scan only the configured MCP server with this exact name. Discovery still runs, then every other server is dropped. |
+| `--url URL` | string | none | Scan a remote MCP server at this URL directly, without reading any config file. Spelled the same as `mcp-auth --url`. |
+| `--server-type {http,sse}` | choice | none | Pin the transport of the targeted server. **Disables transport probing** — exactly the given URL and transport are contacted, once. |
+
+```bash
+# one configured server, by name
+snyk-agent-scan scan --server MY_SERVER
+
+# an ad-hoc URL, no config entry needed, no probing
+snyk-agent-scan scan --url https://api.snyk.io/mcp-server/mcp --server-type http
+
+# override a wrong transport recorded in a config file
+snyk-agent-scan scan --server MY_SERVER --server-type sse
+```
+
+**Behavior:**
+
+- Either flag **skips skills entirely**, the same as passing `--no-skills`.
+- `--url` and `--server` combine: `--url` is the target, `--server` is only the display name. This mirrors `mcp-auth`.
+- Without `--server-type`, a remote server is probed across six transport/URL combinations (`http` and `sse` × the base URL, `/mcp`, and `/sse`). With it, the URL is used verbatim — no `/mcp` or `/sse` suffix is appended or stripped — and a failure raises the underlying error instead of an `ExceptionGroup`.
+- `--server-type` requires `--server` or `--url` (exit code 2), and cannot be applied to a stdio server (exit code 2).
+- A `--server NAME` that matches nothing lists the discovered server names and exits 1.
+- Credentials are unaffected: stored OAuth tokens are looked up by normalized server URL, so a token obtained via `mcp-auth` resolves regardless of which form you use.
+
+### Scanning a server by package or URL without a config
+
+`scan`, `inspect`, and `evo` also accept a **prefixed positional** that describes a server directly, bypassing config discovery:
+
+| Prefix | Example | Resolves to |
+| --- | --- | --- |
+| `streamable-https:` | `streamable-https:api.snyk.io/mcp-server/mcp` | remote server at `https://…` |
+| `streamable-http:` | `streamable-http:localhost:3000/mcp` | remote server at `http://…` |
+| `sse:` | `sse:https://example.com/sse` | remote server, SSE transport |
+| `npm:` | `npm:some-mcp-server@1.2.3` | `npx -y some-mcp-server@1.2.3` |
+| `pypi:` | `pypi:some-mcp-server@1.2.3` | `uvx some-mcp-server@1.2.3` |
+| `oci:` | `oci:ghcr.io/org/image:tag` | `docker run -i --rm …` |
+| `nuget:`, `mcpb:` | — | reserved |
+
+Omit the scheme for `streamable-https:` / `streamable-http:` — the prefix supplies it. These forms leave the transport unset, so probing still applies; use `--url` with `--server-type` when you want a single, exact attempt.
+
 ### Analysis and upload
 
 | Flag | Type | Default | Description |

docs/scanning.md

@@ -65,6 +65,57 @@ Agent applications, skills, tool names, and descriptions are shared with Snyk. R
 
 Discovered client information, MCP server configurations and signatures, and skill files are shared with Snyk. Secrets in configuration values and text are redacted before transmission. Results use scored risk indicators; see the [risk reference](risks.md). Operational errors remain separate in the [failure code reference](failure-codes.md).
 
+## Env var placeholders in stdio server configs
+
+A stdio MCP server's `env` block in a client config (`.claude.json`, `.mcp.json`,
+etc.) can reference `${NAME}` to pull a value from whichever environment is
+running the scan, instead of hardcoding a secret in the config file:
+
+```json
+{
+  "mcpServers": {
+    "my-server": {
+      "command": "npx",
+      "args": ["-y", "mcp-remote@0.3.0", "https://example.com/mcp", "--header", "Authorization:${AUTH_HEADER}"],
+      "env": { "AUTH_HEADER": "${AUTH_HEADER}" }
+    }
+  }
+}
+```
+
+`${AUTH_HEADER}` in the `env` value is substituted from the scanning
+process's own environment variable of the same name at the moment the
+server is spawned. Note the child process's env-var *key* (the left side,
+`"AUTH_HEADER"` here) doesn't have to match the placeholder name on the
+right — `"env": {"FOO_TOKEN": "${MY_SECRET}"}` is equally valid. If the
+referenced variable isn't set when the scan runs, the placeholder is left
+unexpanded and a warning is logged (visible with `--verbose`) rather than
+silently connecting with a blank credential. There is no escape syntax for
+a literal `${NAME}` today: if an `env` value happens to legitimately
+contain well-formed `${NAME}` text that isn't meant as a placeholder, it
+will still be substituted whenever `NAME` happens to be set in the
+scanning environment.
+
+This only applies to `env` values. A value inside `args` (as in the
+`mcp-remote` example above) is *not* expanded by Agent Scan itself --
+`mcp-remote` and similar wrapper commands already resolve `${VAR}`
+references in their own arguments from their own process environment, so
+once the variable is present in the spawned process's environment (via the
+`env` block), that resolution happens downstream, in the wrapper.
+
+Because a config's `env` block can reference `${NAME}` for any variable
+name, a discovered (and possibly untrusted or tampered) config can use
+this feature to pull *any* variable out of the scanning user's shell --
+for example `"env": {"X": "${SNYK_TOKEN}"}`, or a reference to
+`AWS_SECRET_ACCESS_KEY`, `GITHUB_TOKEN`, or similar. This is a natural
+consequence of the feature rather than itself a bug: starting a stdio MCP
+server already requires explicit consent (or `--dangerously-run-mcp-servers`),
+and spawning a server is arbitrary code execution regardless of whether its
+`env` block references any placeholders. To make this visible, the
+interactive consent prompt (shown unless `--dangerously-run-mcp-servers` is
+passed) displays which environment variable names a server's `env` block
+will pull from your shell, so you can review them before the server starts.
+
 ## CLI Usage
 
 The command structure and most options are shared by both versions:

pyproject.toml

@@ -22,6 +22,7 @@ dependencies = [
     "mcp[cli]==1.28.1",
     "regex>=2026.2.19",
     "detect-secrets>=1.5.0",
+    "cryptography>=42.0.0",
     # TOML decoder for the Codex discoverer on Python 3.10 (stdlib ``tomllib`` is
     # 3.11+). Declared directly so the ``tomli`` fallback in ``agents/codex.py`` is
     # guaranteed under ``requires-python>=3.10`` rather than relying on a transitive.

src/agent_scan/cli.py

@@ -37,11 +37,21 @@
     InspectArgs,
     PushArgs,
     discover_clients_to_inspect,
+    discover_servers_by_name,
+    filter_clients_to_server,
     inspect_analyze_push_pipeline,
     inspect_pipeline,
+    single_remote_client_to_inspect,
 )
 from agent_scan.printer import print_inspected_machine, print_scan_response
-from agent_scan.utils import ensure_unicode_console, get_hostname, get_push_key, parse_headers, suppress_stdout
+from agent_scan.utils import (
+    ensure_unicode_console,
+    get_hostname,
+    get_push_key,
+    get_username,
+    parse_headers,
+    suppress_stdout,
+)
 from agent_scan.version import version_info
 
 # Configure logging to suppress all output by default
@@ -662,6 +672,66 @@ def add_ignore_failure_codes_argument(parser) -> None:
     )
 
 
+def add_target_arguments(parser, *, positional: bool, include_type: bool):
+    """Add the "which server" arguments shared by scan/inspect and mcp-auth.
+
+    ``mcp-auth`` takes the server name as a positional (its historical CLI);
+    ``scan``/``inspect`` must use a flag because they already own a greedy
+    ``files`` positional. ``--url`` is spelled identically everywhere so the
+    two commands share one vocabulary.
+    """
+    if positional:
+        parser.add_argument("server", nargs="?", help="Name of the MCP server (as configured) to authenticate")
+    else:
+        parser.add_argument(
+            "--server",
+            help="Scan only the MCP server with this name, as configured (skips every other server and all skills)",
+            metavar="NAME",
+        )
+    parser.add_argument(
+        "--url",
+        help=(
+            "Target a remote MCP server by URL directly, without reading any config file"
+            if not positional
+            else "Authenticate a remote MCP server by URL directly"
+        ),
+    )
+    if include_type:
+        parser.add_argument(
+            "--server-type",
+            choices=["http", "sse"],
+            default=None,
+            help=(
+                "Pin the transport of the targeted server. Disables transport probing, "
+                "so exactly the given URL and transport are contacted, once."
+            ),
+        )
+
+
+def print_server_not_found(server_name: str, discovered: dict, *, remote_only: bool = False) -> None:
+    """Report a --server / mcp-auth name that matched nothing, listing what exists."""
+    kind = "remote MCP server" if remote_only else "MCP server"
+    rich.print(f"[bold red]No {kind} named '{server_name}' found.[/bold red]")
+    if discovered:
+        rich.print(f"Discovered {'remote ' if remote_only else ''}servers: {', '.join(sorted(discovered))}")
+    else:
+        rich.print(f"No {kind}s were discovered on this machine.")
+
+
+def _target_is_remote(clients_to_inspect: list) -> bool:
+    """True when every matched entry in the narrowed plan is a remote server."""
+    from agent_scan.models import RemoteServer
+
+    for client in clients_to_inspect:
+        for entries in client.mcp_configs.values():
+            if not isinstance(entries, list):
+                continue
+            for _name, config in entries:
+                if not isinstance(config, RemoteServer):
+                    return False
+    return True
+
+
 def setup_scan_parser(scan_parser, add_files=True, add_ci_ignore_options=True, add_show_full_discovery_option=True):
     if add_files:
         scan_parser.add_argument(
@@ -871,6 +941,7 @@ def main():
         ),
     )
     setup_scan_parser(scan_parser)
+    add_target_arguments(scan_parser, positional=False, include_type=True)
 
     # INSPECT command
     inspect_parser = subparsers.add_parser(
@@ -891,6 +962,7 @@ def main():
         help="Configuration files to inspect (default: known MCP config locations)",
         metavar="CONFIG_FILE",
     )
+    add_target_arguments(inspect_parser, positional=False, include_type=True)
 
     # HELP command
     help_parser = subparsers.add_parser(  # noqa: F841
@@ -905,6 +977,16 @@ def main():
     # use the same parser as scan
     setup_scan_parser(evo_parser, add_ci_ignore_options=False, add_show_full_discovery_option=False)
 
+    # MCP-AUTH command: interactively authenticate an OAuth-protected remote server
+    mcp_auth_parser = subparsers.add_parser(
+        "mcp-auth", help="Authenticate an OAuth-protected remote MCP server so scans can use it"
+    )
+    setup_scan_parser(mcp_auth_parser, add_files=False)
+    add_target_arguments(mcp_auth_parser, positional=True, include_type=False)
+    mcp_auth_parser.add_argument(
+        "--all-unauthenticated", action="store_true", help="Authenticate every discovered remote MCP server"
+    )
+
     # GUARD command
     guard_parser = subparsers.add_parser(
         "guard",
@@ -1032,6 +1114,8 @@ def main():
     elif args.command == "evo":
         asyncio.run(evo(args))
         sys.exit(0)
+    elif args.command == "mcp-auth":
+        sys.exit(asyncio.run(mcp_auth(args)))
     elif args.command == "guard":
         from agent_scan.guard import run_guard
 
@@ -1108,6 +1192,69 @@ def _should_show_analysis_results(args) -> bool:
     )
 
 
+async def mcp_auth(args) -> int:
+    """Interactively authenticate an OAuth-protected remote MCP server.
+
+    Runs the browser OAuth flow and persists the token to the local store, so
+    subsequent (unattended) scans use and refresh it. This is the only command
+    that performs an interactive authorization; the scan path never does.
+
+    Returns an exit status: 0 if every requested target authenticated
+    successfully, 1 otherwise (including invalid/missing target selection), so
+    scripts can detect failure instead of always seeing a zero exit.
+    """
+    from urllib.parse import urlparse
+
+    from agent_scan.oauth_flow import authenticate_server
+    from agent_scan.oauth_store import OAuthTokenStore
+
+    store = OAuthTokenStore()
+    url_arg = getattr(args, "url", None)
+    server_arg = getattr(args, "server", None)
+    all_unauth = getattr(args, "all_unauthenticated", False)
+
+    targets: list[tuple[str, str]] = []
+    if url_arg:
+        name = server_arg or urlparse(url_arg).hostname or url_arg
+        targets = [(name, url_arg)]
+    else:
+        # Discover remote MCP servers from this machine's agent configs.
+        inspect_args = InspectArgs(
+            timeout=getattr(args, "server_timeout", 10),
+            tokens=[],
+            paths=[],
+            all_users=getattr(args, "scan_all_users", False),
+            scan_skills=False,
+        )
+        discovered = await discover_servers_by_name(inspect_args, remote_only=True)
+        remote: dict[str, str] = {name: server.url for name, server in discovered.items()}
+        if all_unauth:
+            targets = list(remote.items())
+        elif server_arg:
+            if server_arg not in remote:
+                print_server_not_found(server_arg, remote, remote_only=True)
+                return 1
+            targets = [(server_arg, remote[server_arg])]
+        else:
+            rich.print("[bold red]Specify a server name, --url <url>, or --all-unauthenticated.[/bold red]")
+            return 1
+
+    if not targets:
+        rich.print("No remote MCP servers to authenticate.")
+        return 1
+
+    all_ok = True
+    for name, url in targets:
+        rich.print(f"\n[bold]Authenticating '{name}'[/bold] ({url}) ...")
+        result = await authenticate_server(url, name, store)
+        if result.ok:
+            rich.print(f"[bold green]{name}: authenticated[/bold green]")
+        else:
+            rich.print(f"[bold red]{name}: authentication failed[/bold red] — {result.message}")
+            all_ok = False
+    return 0 if all_ok else 1
+
+
 async def run_scan(args, mode: Literal["scan", "inspect"] = "scan") -> ScanResponse | list[InspectedPath]:
     """
     Run the scan or inspect flow through their shared discovery and consent setup.
@@ -1131,17 +1278,33 @@ async def run_scan(args, mode: Literal["scan", "inspect"] = "scan") -> ScanRespo
     server_timeout: int = args.server_timeout if hasattr(args, "server_timeout") else 10
     files: list[str] | None = args.files if hasattr(args, "files") else None
     scan_skills: bool = hasattr(args, "skills") and args.skills
+
     tokens: list[TokenAndClientInfo] = []
     if hasattr(args, "mcp_oauth_tokens_path") and args.mcp_oauth_tokens_path:
         with open(args.mcp_oauth_tokens_path) as f:
             tokens = TokenAndClientInfoList.model_validate_json(f.read()).root
 
+    # Single-server targeting. --url addresses a server directly (no discovery);
+    # --server narrows discovery to one configured entry. Either way skills are
+    # irrelevant, and --server-type pins the transport so nothing is probed.
+    target_name: str | None = getattr(args, "server", None)
+    target_url: str | None = getattr(args, "url", None)
+    # argparse restricts --server-type to choices=["http", "sse"] (see
+    # add_target_arguments), so the runtime value always matches the literal.
+    target_type = cast('Literal["sse", "http"] | None', getattr(args, "server_type", None))
+    if target_type and not (target_name or target_url):
+        rich.print("[bold red]--server-type requires --server <name> or --url <url>.[/bold red]")
+        sys.exit(2)
+    if target_name or target_url:
+        scan_skills = False
+
     inspect_args = InspectArgs(
         timeout=server_timeout,
         tokens=tokens,
         paths=files,
         all_users=scan_all_users,
         scan_skills=scan_skills,
+        probe_transports=not bool(target_type),
     )
 
     # Resolve the MCP server IO flag and the consent flag.
@@ -1154,7 +1317,27 @@ async def run_scan(args, mode: Literal["scan", "inspect"] = "scan") -> ScanRespo
     dangerously_run_mcp_servers: bool = bool(getattr(args, "dangerously_run_mcp_servers", False))
 
     # Step 1: Discover everything we would inspect without starting any server.
-    clients_to_inspect, unresolved_paths, scanned_usernames = await discover_clients_to_inspect(inspect_args)
+    if target_url:
+        # Addressed directly: skip discovery entirely.
+        clients_to_inspect = [single_remote_client_to_inspect(target_name, target_url, target_type)]
+        unresolved_paths: list[InspectedPath] = []
+        scanned_usernames = [get_username()]
+    else:
+        clients_to_inspect, unresolved_paths, scanned_usernames = await discover_clients_to_inspect(inspect_args)
+        if target_name:
+            clients_to_inspect = filter_clients_to_server(clients_to_inspect, target_name, target_type)
+            if not clients_to_inspect:
+                discovered = await discover_servers_by_name(inspect_args)
+                print_server_not_found(target_name, discovered)
+                sys.exit(1)
+            if target_type and not _target_is_remote(clients_to_inspect):
+                rich.print(
+                    f"[bold red]'{target_name}' is a stdio server; --server-type only applies to remote servers."
+                    "[/bold red]"
+                )
+                sys.exit(2)
+            # A named target is one server: nothing else should be reported.
+            unresolved_paths = []
 
     # Collect consent when applicable; otherwise show the
     # dangerous-flag banner to users at the terminal. Silent