mirror of
https://github.com/github/spec-kit.git
synced 2026-08-03 06:26:30 +08:00
* fix(auth): resolve az via shutil.which so azure-cli token works on Windows
AzureDevOpsAuth._acquire_via_az_cli runs subprocess.run with a bare "az".
On Windows the Azure CLI is installed as az.cmd, and subprocess.run calls
CreateProcess, which does not consult PATHEXT -- so a bare "az" fails with
WinError 2 even after `az login`, and azure-cli token acquisition silently
returns None (the OSError is swallowed).
Resolve the executable with shutil.which("az") (which honors PATHEXT) before
the call, mirroring the maintainer's own fix in integrations/base.py for the
same CreateProcess/.cmd issue. `or "az"` preserves prior behavior (and the
existing not-installed OSError path) when az is absent. POSIX is unaffected.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* fix(auth): require an absolute az path so the CWD cannot hijack the lookup
Self-review catch on my own change: resolving with a bare
`shutil.which("az") or "az"` widened an execution surface. On Windows
shutil.which prepends the CURRENT DIRECTORY to the search path (unless
NoDefaultCurrentDirectoryInExePath is set) AND honors PATHEXT, so a stray
.\az.cmd / .\az.bat in the working directory resolves ahead of the real Azure
CLI -- for a credential operation. Verified: with the real az scrubbed from
PATH, shutil.which("az") returns '.\az.CMD'.
Accept the resolution only when it is absolute; otherwise fall back to the bare
"az" (which also preserves the existing not-installed OSError path). A
legitimate install always resolves absolutely, so the Windows .cmd fix this PR
exists for is unaffected. The not-installed and PATHEXT tests are extended with
relative-result cases, all of which fail before this commit.
Note: integrations/base.py resolves executables the same way; hardening that
shared path is a separate concern and is left untouched here.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* test(auth): build the mocked az path with the host's path rules
Fixes the macOS CI failure. The test hardcoded a Windows absolute path, but the
production code calls os.path.isabs() -- on POSIX runners "C:\Program
Files\..." reads as RELATIVE, so the fallback branch ran and argv[0] was "az"
instead of the resolved path.
Construct the path with os.path.join(os.path.abspath(os.sep), ...) so it is
absolute under the host's rules, and assert against that value. The fallback
test's inputs (".\az.CMD", "az.cmd", "./az") are relative under both ntpath
and posixpath, so they were already portable.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1394 lines
58 KiB
Python
1394 lines
58 KiB
Python
"""Tests for the authentication provider registry and config-driven HTTP helpers.
|
||
|
||
Covers:
|
||
- Config loading (auth.json parsing, validation, permission warning)
|
||
- Registry mechanics (_register, get_provider, duplicate/empty-key guards)
|
||
- GitHubAuth — bearer headers
|
||
- AzureDevOpsAuth — basic-pat, bearer, azure-cli, azure-ad headers
|
||
- Host matching (find_entries_for_url)
|
||
- open_url — config-driven auth with fallthrough and redirect stripping
|
||
- build_request — single-shot request construction
|
||
- _fetch_latest_release_tag() delegation
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import io
|
||
import json
|
||
import os
|
||
|
||
import pytest
|
||
|
||
from specify_cli.authentication import AUTH_REGISTRY, _register, get_provider
|
||
from specify_cli.authentication.azure_devops import AzureDevOpsAuth
|
||
from specify_cli.authentication.base import AuthProvider
|
||
from specify_cli.authentication.config import (
|
||
AuthConfigEntry,
|
||
find_entries_for_url,
|
||
load_auth_config,
|
||
)
|
||
from specify_cli.authentication.github import GitHubAuth
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _github_entry(token_env: str = "GH_TOKEN", token: str | None = None) -> AuthConfigEntry:
|
||
"""Build a standard GitHub config entry."""
|
||
return AuthConfigEntry(
|
||
hosts=("github.com", "api.github.com", "raw.githubusercontent.com", "codeload.github.com"),
|
||
provider="github",
|
||
auth="bearer",
|
||
token=token,
|
||
token_env=token_env if token is None else None,
|
||
)
|
||
|
||
|
||
def _ado_basic_entry(token_env: str = "AZURE_DEVOPS_PAT") -> AuthConfigEntry:
|
||
"""Build an ADO basic-pat config entry."""
|
||
return AuthConfigEntry(
|
||
hosts=("dev.azure.com",),
|
||
provider="azure-devops",
|
||
auth="basic-pat",
|
||
token_env=token_env,
|
||
)
|
||
|
||
|
||
class _StubProvider(AuthProvider):
|
||
"""Minimal concrete provider for registry mechanics tests."""
|
||
|
||
key = "stub-provider"
|
||
supported_auth_schemes = ("bearer",)
|
||
|
||
def auth_headers(self, token: str, auth_scheme: str) -> dict[str, str]:
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Config loading
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestLoadAuthConfig:
|
||
def test_missing_file_returns_empty(self, tmp_path):
|
||
assert load_auth_config(tmp_path / "nonexistent.json") == []
|
||
|
||
def test_valid_github_config(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{
|
||
"hosts": ["github.com"],
|
||
"provider": "github",
|
||
"auth": "bearer",
|
||
"token_env": "GH_TOKEN",
|
||
}]
|
||
}))
|
||
entries = load_auth_config(cfg)
|
||
assert len(entries) == 1
|
||
assert entries[0].provider == "github"
|
||
assert entries[0].auth == "bearer"
|
||
assert entries[0].token_env == "GH_TOKEN"
|
||
|
||
def test_padded_token_env_is_normalized_and_resolves(self, tmp_path, monkeypatch):
|
||
# token_env is validated on its stripped form but was stored raw, so a
|
||
# padded env-var name passed validation yet broke the verbatim
|
||
# os.environ.get() lookup — resolve_token silently returned None.
|
||
monkeypatch.setenv("GH_TOKEN", "secret-tok")
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{
|
||
"hosts": ["github.com"],
|
||
"provider": "github",
|
||
"auth": "bearer",
|
||
"token_env": " GH_TOKEN ",
|
||
}]
|
||
}))
|
||
entries = load_auth_config(cfg)
|
||
assert len(entries) == 1
|
||
# Stored normalized (matching how hosts are normalized), ...
|
||
assert entries[0].token_env == "GH_TOKEN"
|
||
# ... so the env lookup finds the token instead of returning None.
|
||
assert GitHubAuth().resolve_token(entries[0]) == "secret-tok"
|
||
|
||
def test_valid_ado_config(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{
|
||
"hosts": ["dev.azure.com"],
|
||
"provider": "azure-devops",
|
||
"auth": "basic-pat",
|
||
"token_env": "AZURE_DEVOPS_PAT",
|
||
}]
|
||
}))
|
||
entries = load_auth_config(cfg)
|
||
assert len(entries) == 1
|
||
assert entries[0].provider == "azure-devops"
|
||
assert entries[0].auth == "basic-pat"
|
||
|
||
def test_inline_token(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{
|
||
"hosts": ["github.com"],
|
||
"provider": "github",
|
||
"auth": "bearer",
|
||
"token": "ghp_inline_token",
|
||
}]
|
||
}))
|
||
entries = load_auth_config(cfg)
|
||
assert entries[0].token == "ghp_inline_token"
|
||
|
||
def test_azure_ad_config(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{
|
||
"hosts": ["dev.azure.com"],
|
||
"provider": "azure-devops",
|
||
"auth": "azure-ad",
|
||
"tenant_id": "tid",
|
||
"client_id": "cid",
|
||
"client_secret_env": "SECRET",
|
||
}]
|
||
}))
|
||
entries = load_auth_config(cfg)
|
||
assert entries[0].auth == "azure-ad"
|
||
assert entries[0].tenant_id == "tid"
|
||
|
||
def test_padded_azure_ad_refs_are_normalized(self, tmp_path):
|
||
# The normalization also covers tenant_id / client_id / client_secret_env
|
||
# (used verbatim in the OAuth token URL/body and os.environ.get). Padded
|
||
# values were validated on their stripped form but stored raw.
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{
|
||
"hosts": ["dev.azure.com"],
|
||
"provider": "azure-devops",
|
||
"auth": "azure-ad",
|
||
"tenant_id": " tid ",
|
||
"client_id": " cid ",
|
||
"client_secret_env": " SECRET ",
|
||
}]
|
||
}))
|
||
entries = load_auth_config(cfg)
|
||
assert entries[0].tenant_id == "tid"
|
||
assert entries[0].client_id == "cid"
|
||
assert entries[0].client_secret_env == "SECRET"
|
||
|
||
def test_azure_cli_config(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{
|
||
"hosts": ["dev.azure.com"],
|
||
"provider": "azure-devops",
|
||
"auth": "azure-cli",
|
||
}]
|
||
}))
|
||
entries = load_auth_config(cfg)
|
||
assert entries[0].auth == "azure-cli"
|
||
|
||
def test_multiple_entries(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [
|
||
{"hosts": ["github.com"], "provider": "github", "auth": "bearer", "token_env": "GH_TOKEN"},
|
||
{"hosts": ["dev.azure.com"], "provider": "azure-devops", "auth": "basic-pat", "token_env": "ADO_PAT"},
|
||
]
|
||
}))
|
||
entries = load_auth_config(cfg)
|
||
assert len(entries) == 2
|
||
|
||
# -- Negative: validation errors --
|
||
|
||
def test_invalid_json_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text("not json")
|
||
with pytest.raises(json.JSONDecodeError):
|
||
load_auth_config(cfg)
|
||
|
||
def test_not_object_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text("[]")
|
||
with pytest.raises(ValueError, match="JSON object"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_missing_providers_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({"foo": "bar"}))
|
||
with pytest.raises(ValueError, match="providers"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_empty_hosts_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{"hosts": [], "provider": "github", "auth": "bearer", "token_env": "X"}]
|
||
}))
|
||
with pytest.raises(ValueError, match="non-empty"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_missing_provider_key_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{"hosts": ["github.com"], "auth": "bearer", "token_env": "X"}]
|
||
}))
|
||
with pytest.raises(ValueError, match="provider"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_unsupported_auth_scheme_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{"hosts": ["github.com"], "provider": "github", "auth": "ntlm", "token_env": "X"}]
|
||
}))
|
||
with pytest.raises(ValueError, match="does not support"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_bearer_without_token_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{"hosts": ["github.com"], "provider": "github", "auth": "bearer"}]
|
||
}))
|
||
with pytest.raises(ValueError, match="token"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_azure_ad_missing_fields_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{
|
||
"hosts": ["dev.azure.com"],
|
||
"provider": "azure-devops",
|
||
"auth": "azure-ad",
|
||
"tenant_id": "tid",
|
||
}]
|
||
}))
|
||
with pytest.raises(ValueError, match="azure-ad"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_unknown_provider_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{"hosts": ["example.com"], "provider": "gitlab", "auth": "bearer", "token_env": "X"}]
|
||
}))
|
||
with pytest.raises(ValueError, match="unknown provider"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_incompatible_provider_scheme_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{
|
||
"hosts": ["github.com"],
|
||
"provider": "github",
|
||
"auth": "basic-pat",
|
||
"token_env": "X",
|
||
}]
|
||
}))
|
||
with pytest.raises(ValueError, match="does not support"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_dangerous_wildcard_host_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{"hosts": ["*github.com"], "provider": "github", "auth": "bearer", "token_env": "X"}]
|
||
}))
|
||
with pytest.raises(ValueError, match="invalid host pattern"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_multi_wildcard_host_raises(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{"hosts": ["*.*.example.com"], "provider": "github", "auth": "bearer", "token_env": "X"}]
|
||
}))
|
||
with pytest.raises(ValueError, match="invalid host pattern"):
|
||
load_auth_config(cfg)
|
||
|
||
def test_valid_star_dot_host_accepted(self, tmp_path):
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{"hosts": ["*.visualstudio.com"], "provider": "azure-devops", "auth": "basic-pat", "token_env": "X"}]
|
||
}))
|
||
entries = load_auth_config(cfg)
|
||
assert entries[0].hosts == ("*.visualstudio.com",)
|
||
|
||
@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits not supported on Windows")
|
||
def test_world_readable_warns(self, tmp_path):
|
||
import stat
|
||
|
||
cfg = tmp_path / "auth.json"
|
||
cfg.write_text(json.dumps({
|
||
"providers": [{"hosts": ["github.com"], "provider": "github", "auth": "bearer", "token_env": "GH_TOKEN"}]
|
||
}))
|
||
cfg.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)
|
||
with pytest.warns(UserWarning, match="readable by group"):
|
||
load_auth_config(cfg)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Host matching
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFindEntriesForUrl:
|
||
def test_exact_match(self):
|
||
entry = _github_entry()
|
||
result = find_entries_for_url("https://github.com/org/repo", [entry])
|
||
assert result == [entry]
|
||
|
||
def test_wildcard_match(self):
|
||
entry = AuthConfigEntry(
|
||
hosts=("*.visualstudio.com",),
|
||
provider="azure-devops",
|
||
auth="basic-pat",
|
||
token_env="ADO_PAT",
|
||
)
|
||
result = find_entries_for_url("https://myorg.visualstudio.com/project", [entry])
|
||
assert result == [entry]
|
||
|
||
def test_no_match_returns_empty(self):
|
||
entry = _github_entry()
|
||
result = find_entries_for_url("https://evil.example.com/file", [entry])
|
||
assert result == []
|
||
|
||
def test_no_match_for_lookalike_host(self):
|
||
entry = _github_entry()
|
||
result = find_entries_for_url("https://github.com.evil.com/file", [entry])
|
||
assert result == []
|
||
|
||
def test_empty_url_returns_empty(self):
|
||
assert find_entries_for_url("", [_github_entry()]) == []
|
||
|
||
@pytest.mark.parametrize(
|
||
"url",
|
||
[
|
||
"https://[::1", # unterminated ipv6 bracket
|
||
"https://[not-an-ip]/file", # bracketed non-ip host
|
||
],
|
||
)
|
||
def test_malformed_url_returns_empty(self, url):
|
||
# A malformed authority makes urlparse/hostname raise ValueError.
|
||
# Since no entry can match such a URL, this must return no matches
|
||
# (like a host-less URL) rather than leaking a raw ValueError out of
|
||
# the shared HTTP client.
|
||
assert find_entries_for_url(url, [_github_entry()]) == []
|
||
|
||
def test_empty_entries_returns_empty(self):
|
||
assert find_entries_for_url("https://github.com/org/repo", []) == []
|
||
|
||
def test_multiple_matches_returned(self):
|
||
e1 = _github_entry(token_env="GH_TOKEN")
|
||
e2 = _github_entry(token_env="GITHUB_TOKEN")
|
||
result = find_entries_for_url("https://github.com/org/repo", [e1, e2])
|
||
assert len(result) == 2
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Registry mechanics
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestAuthRegistry:
|
||
def test_github_registered(self):
|
||
assert "github" in AUTH_REGISTRY
|
||
|
||
def test_azure_devops_registered(self):
|
||
assert "azure-devops" in AUTH_REGISTRY
|
||
|
||
def test_get_provider_returns_github(self):
|
||
assert isinstance(get_provider("github"), GitHubAuth)
|
||
|
||
def test_get_provider_returns_azure_devops(self):
|
||
assert isinstance(get_provider("azure-devops"), AzureDevOpsAuth)
|
||
|
||
def test_get_provider_unknown_returns_none(self):
|
||
assert get_provider("does-not-exist") is None
|
||
|
||
def test_register_duplicate_raises_key_error(self):
|
||
class _UniqueStub(_StubProvider):
|
||
key = "__test_duplicate__"
|
||
|
||
try:
|
||
_register(_UniqueStub())
|
||
with pytest.raises(KeyError, match="already registered"):
|
||
_register(_UniqueStub())
|
||
finally:
|
||
AUTH_REGISTRY.pop("__test_duplicate__", None)
|
||
|
||
def test_register_empty_key_raises_value_error(self):
|
||
class _EmptyKey(_StubProvider):
|
||
key = ""
|
||
|
||
with pytest.raises(ValueError, match="empty key"):
|
||
_register(_EmptyKey())
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GitHubAuth
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGitHubAuth:
|
||
def test_bearer_headers(self):
|
||
assert GitHubAuth().auth_headers("my-token", "bearer") == {"Authorization": "Bearer my-token"}
|
||
|
||
def test_unsupported_scheme_raises(self):
|
||
with pytest.raises(ValueError, match="basic-pat"):
|
||
GitHubAuth().auth_headers("tok", "basic-pat")
|
||
|
||
def test_resolve_token_from_env(self, monkeypatch):
|
||
monkeypatch.setenv("GH_TOKEN", "env-token")
|
||
assert GitHubAuth().resolve_token(_github_entry()) == "env-token"
|
||
|
||
def test_resolve_token_inline(self):
|
||
assert GitHubAuth().resolve_token(_github_entry(token="inline-tok")) == "inline-tok"
|
||
|
||
def test_resolve_token_strips_whitespace(self, monkeypatch):
|
||
monkeypatch.setenv("GH_TOKEN", " my-token ")
|
||
assert GitHubAuth().resolve_token(_github_entry()) == "my-token"
|
||
|
||
def test_resolve_token_empty_env_returns_none(self, monkeypatch):
|
||
monkeypatch.setenv("GH_TOKEN", " ")
|
||
assert GitHubAuth().resolve_token(_github_entry()) is None
|
||
|
||
def test_resolve_token_missing_env_returns_none(self, monkeypatch):
|
||
monkeypatch.delenv("GH_TOKEN", raising=False)
|
||
assert GitHubAuth().resolve_token(_github_entry()) is None
|
||
|
||
def test_key(self):
|
||
assert GitHubAuth.key == "github"
|
||
|
||
def test_supported_schemes(self):
|
||
assert GitHubAuth.supported_auth_schemes == ("bearer",)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# AzureDevOpsAuth
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestAzureDevOpsAuth:
|
||
def test_basic_pat_headers(self):
|
||
headers = AzureDevOpsAuth().auth_headers("my-pat", "basic-pat")
|
||
encoded = base64.b64encode(b":my-pat").decode("ascii")
|
||
assert headers == {"Authorization": f"Basic {encoded}"}
|
||
|
||
def test_basic_pat_format(self):
|
||
header = AzureDevOpsAuth().auth_headers("test-pat", "basic-pat")["Authorization"]
|
||
raw = base64.b64decode(header[len("Basic "):]).decode("ascii")
|
||
assert raw == ":test-pat"
|
||
|
||
def test_bearer_headers(self):
|
||
assert AzureDevOpsAuth().auth_headers("tok", "bearer") == {"Authorization": "Bearer tok"}
|
||
|
||
def test_azure_cli_headers(self):
|
||
assert AzureDevOpsAuth().auth_headers("tok", "azure-cli") == {"Authorization": "Bearer tok"}
|
||
|
||
def test_azure_ad_headers(self):
|
||
assert AzureDevOpsAuth().auth_headers("tok", "azure-ad") == {"Authorization": "Bearer tok"}
|
||
|
||
def test_unsupported_scheme_raises(self):
|
||
with pytest.raises(ValueError):
|
||
AzureDevOpsAuth().auth_headers("tok", "ntlm")
|
||
|
||
def test_resolve_token_basic_pat(self, monkeypatch):
|
||
monkeypatch.setenv("AZURE_DEVOPS_PAT", "my-pat")
|
||
assert AzureDevOpsAuth().resolve_token(_ado_basic_entry()) == "my-pat"
|
||
|
||
def test_resolve_token_strips_whitespace(self, monkeypatch):
|
||
monkeypatch.setenv("AZURE_DEVOPS_PAT", " my-pat ")
|
||
assert AzureDevOpsAuth().resolve_token(_ado_basic_entry()) == "my-pat"
|
||
|
||
def test_resolve_token_missing_returns_none(self, monkeypatch):
|
||
monkeypatch.delenv("AZURE_DEVOPS_PAT", raising=False)
|
||
assert AzureDevOpsAuth().resolve_token(_ado_basic_entry()) is None
|
||
|
||
def test_key(self):
|
||
assert AzureDevOpsAuth.key == "azure-devops"
|
||
|
||
def test_supported_schemes(self):
|
||
schemes = AzureDevOpsAuth.supported_auth_schemes
|
||
assert "basic-pat" in schemes
|
||
assert "bearer" in schemes
|
||
assert "azure-cli" in schemes
|
||
assert "azure-ad" in schemes
|
||
|
||
def test_resolve_token_azure_cli_success(self):
|
||
"""azure-cli acquires token via az CLI."""
|
||
from unittest.mock import patch, MagicMock
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-cli",
|
||
)
|
||
result = MagicMock()
|
||
result.returncode = 0
|
||
result.stdout = '{"accessToken": "cli-acquired-token"}'
|
||
with patch("specify_cli.authentication.azure_devops.subprocess.run", return_value=result):
|
||
assert AzureDevOpsAuth().resolve_token(entry) == "cli-acquired-token"
|
||
|
||
def test_resolve_token_azure_cli_failure_returns_none(self):
|
||
"""azure-cli returns None when az CLI fails."""
|
||
from unittest.mock import patch, MagicMock
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-cli",
|
||
)
|
||
result = MagicMock()
|
||
result.returncode = 1
|
||
result.stdout = ""
|
||
with patch("specify_cli.authentication.azure_devops.subprocess.run", return_value=result):
|
||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||
|
||
def test_resolve_token_azure_cli_resolves_executable(self):
|
||
"""The az executable is resolved via shutil.which before invocation, so
|
||
the .cmd/.bat shim on Windows (CreateProcess ignores PATHEXT) is used."""
|
||
from unittest.mock import patch, MagicMock
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-cli",
|
||
)
|
||
# Build the absolute path with the HOST's rules: the production code
|
||
# calls os.path.isabs(), so a hardcoded Windows path would read as
|
||
# RELATIVE on POSIX runners and silently exercise the fallback branch
|
||
# instead of the one under test.
|
||
resolved_path = os.path.join(os.path.abspath(os.sep), "opt", "az", "az.CMD")
|
||
assert os.path.isabs(resolved_path)
|
||
result = MagicMock()
|
||
result.returncode = 0
|
||
result.stdout = '{"accessToken": "tok"}'
|
||
with patch(
|
||
"specify_cli.authentication.azure_devops.shutil.which",
|
||
return_value=resolved_path,
|
||
), patch(
|
||
"specify_cli.authentication.azure_devops.subprocess.run",
|
||
return_value=result,
|
||
) as run:
|
||
assert AzureDevOpsAuth().resolve_token(entry) == "tok"
|
||
argv = run.call_args.args[0]
|
||
assert argv[0] == resolved_path
|
||
assert argv[1:4] == ["account", "get-access-token", "--resource"]
|
||
|
||
@pytest.mark.parametrize("which_result", [None, r".\az.CMD", "az.cmd", "./az"])
|
||
def test_resolve_token_azure_cli_falls_back_to_bare_az(self, which_result):
|
||
"""Fall back to the bare "az" when shutil.which finds nothing OR returns a
|
||
NON-ABSOLUTE path. On Windows shutil.which searches the current directory
|
||
first, so a stray .\\az.cmd must never be executed for a credential
|
||
operation; the bare name also preserves the not-installed OSError path."""
|
||
from unittest.mock import patch, MagicMock
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-cli",
|
||
)
|
||
result = MagicMock()
|
||
result.returncode = 0
|
||
result.stdout = '{"accessToken": "tok"}'
|
||
with patch(
|
||
"specify_cli.authentication.azure_devops.shutil.which",
|
||
return_value=which_result,
|
||
), patch(
|
||
"specify_cli.authentication.azure_devops.subprocess.run",
|
||
return_value=result,
|
||
) as run:
|
||
assert AzureDevOpsAuth().resolve_token(entry) == "tok"
|
||
assert run.call_args.args[0][0] == "az", which_result
|
||
|
||
def test_resolve_token_azure_cli_not_installed_returns_none(self):
|
||
"""azure-cli returns None when az is not installed."""
|
||
from unittest.mock import patch
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-cli",
|
||
)
|
||
with patch("specify_cli.authentication.azure_devops.subprocess.run", side_effect=OSError("not found")):
|
||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||
|
||
def test_resolve_token_azure_cli_undecodable_output_returns_none(self):
|
||
"""Undecodable az output returns None, not a crash. With text=True,
|
||
subprocess.run decodes stdout with the locale encoding and raises
|
||
UnicodeDecodeError (not a JSONDecodeError) when it can't — the helper's
|
||
contract is to return None on any failure."""
|
||
from unittest.mock import patch
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-cli",
|
||
)
|
||
boom = UnicodeDecodeError("utf-8", b"\xff\xfe", 0, 1, "invalid start byte")
|
||
with patch("specify_cli.authentication.azure_devops.subprocess.run", side_effect=boom):
|
||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||
|
||
@pytest.mark.parametrize("payload", [[], {"accessToken": None}, {"accessToken": 123}])
|
||
def test_resolve_token_azure_cli_unexpected_json_shape_returns_none(
|
||
self, payload
|
||
):
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-cli",
|
||
)
|
||
result = MagicMock(returncode=0, stdout=json.dumps(payload))
|
||
|
||
with patch(
|
||
"specify_cli.authentication.azure_devops.subprocess.run",
|
||
return_value=result,
|
||
):
|
||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||
|
||
def test_resolve_token_azure_ad_success(self, monkeypatch):
|
||
"""azure-ad acquires token via OAuth2 client credentials."""
|
||
from unittest.mock import patch, MagicMock
|
||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||
)
|
||
mock_resp = MagicMock()
|
||
mock_resp.read.side_effect = io.BytesIO(b'{"access_token": "ad-acquired-token"}').read
|
||
mock_resp.__enter__ = lambda s: s
|
||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||
# The token request goes through a strict-redirect opener (so a 307/308
|
||
# cannot forward the client_secret body to a non-HTTPS host), not bare
|
||
# urlopen; patch the opener it builds.
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.return_value = mock_resp
|
||
with patch("urllib.request.build_opener", return_value=mock_opener):
|
||
assert AzureDevOpsAuth().resolve_token(entry) == "ad-acquired-token"
|
||
|
||
def test_resolve_token_azure_ad_missing_secret_returns_none(self, monkeypatch):
|
||
"""azure-ad returns None when client secret env var is missing."""
|
||
monkeypatch.delenv("MY_SECRET", raising=False)
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||
)
|
||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||
|
||
def test_resolve_token_azure_ad_network_error_returns_none(self, monkeypatch):
|
||
"""azure-ad returns None on network errors."""
|
||
import urllib.error
|
||
from unittest.mock import MagicMock, patch
|
||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||
)
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = urllib.error.URLError("connection refused")
|
||
with patch("urllib.request.build_opener", return_value=mock_opener):
|
||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||
|
||
@pytest.mark.parametrize(
|
||
("status", "reason"),
|
||
[(307, "Temporary Redirect"), (308, "Permanent Redirect")],
|
||
)
|
||
def test_resolve_token_azure_ad_rejects_https_redirect(
|
||
self, monkeypatch, status, reason
|
||
):
|
||
"""The client-secret POST must never be redirected to another host."""
|
||
import urllib.error
|
||
from unittest.mock import MagicMock, patch
|
||
from urllib.request import Request
|
||
|
||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||
)
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = urllib.error.URLError("stop after setup")
|
||
|
||
with patch("urllib.request.build_opener", return_value=mock_opener) as build_opener:
|
||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||
|
||
redirect_handler = build_opener.call_args.args[0]
|
||
request = Request(
|
||
"https://login.microsoftonline.com/tid/oauth2/v2.0/token",
|
||
data=b"grant_type=client_credentials&client_secret=secret-value",
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
)
|
||
assert request.get_method() == "POST"
|
||
assert b"client_secret=secret-value" in request.data
|
||
with pytest.raises(urllib.error.URLError, match="must not be redirected"):
|
||
redirect_handler.redirect_request(
|
||
request,
|
||
io.BytesIO(b""),
|
||
status,
|
||
reason,
|
||
{},
|
||
"https://evil.example/token",
|
||
)
|
||
|
||
def test_resolve_token_azure_ad_oversized_response_returns_none(
|
||
self, monkeypatch
|
||
):
|
||
"""Oversized token metadata is rejected before JSON parsing."""
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from specify_cli._download_security import MAX_JSON_METADATA_BYTES
|
||
|
||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||
)
|
||
mock_resp = MagicMock()
|
||
mock_resp.read.side_effect = io.BytesIO(
|
||
b"x" * (MAX_JSON_METADATA_BYTES + 1)
|
||
).read
|
||
mock_resp.__enter__ = lambda s: s
|
||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.return_value = mock_resp
|
||
|
||
with patch("urllib.request.build_opener", return_value=mock_opener), patch(
|
||
"specify_cli.authentication.azure_devops._json.loads",
|
||
side_effect=AssertionError("oversized body must not be parsed"),
|
||
):
|
||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||
|
||
@pytest.mark.parametrize("payload", [[], {"access_token": None}, {"access_token": 123}])
|
||
def test_resolve_token_azure_ad_unexpected_json_shape_returns_none(
|
||
self, monkeypatch, payload
|
||
):
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||
)
|
||
mock_resp = MagicMock()
|
||
mock_resp.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read
|
||
mock_resp.__enter__ = lambda s: s
|
||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.return_value = mock_resp
|
||
|
||
with patch("urllib.request.build_opener", return_value=mock_opener):
|
||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||
|
||
def test_resolve_token_azure_ad_invalid_utf8_returns_none(self, monkeypatch):
|
||
"""azure-ad returns None when the token response is not valid UTF-8."""
|
||
from unittest.mock import MagicMock, patch
|
||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||
entry = AuthConfigEntry(
|
||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||
)
|
||
mock_resp = MagicMock()
|
||
mock_resp.read.side_effect = io.BytesIO(b"\xff").read
|
||
mock_resp.__enter__ = lambda s: s
|
||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.return_value = mock_resp
|
||
|
||
with patch("urllib.request.build_opener", return_value=mock_opener):
|
||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# open_url / build_request — positive tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestAuthenticatedHttp:
|
||
def _set_config(self, monkeypatch, entries):
|
||
from specify_cli.authentication import http as _mod
|
||
monkeypatch.setattr(_mod, "_config_override", entries)
|
||
|
||
def test_build_request_attaches_auth_for_matching_host(self, monkeypatch):
|
||
from specify_cli.authentication.http import build_request
|
||
monkeypatch.setenv("GH_TOKEN", "my-token")
|
||
self._set_config(monkeypatch, [_github_entry()])
|
||
req = build_request("https://github.com/org/repo")
|
||
assert req.get_header("Authorization") == "Bearer my-token"
|
||
|
||
def test_build_request_no_auth_for_non_matching_host(self, monkeypatch):
|
||
from specify_cli.authentication.http import build_request
|
||
monkeypatch.setenv("GH_TOKEN", "my-token")
|
||
self._set_config(monkeypatch, [_github_entry()])
|
||
req = build_request("https://evil.example.com/file")
|
||
assert "Authorization" not in req.headers
|
||
|
||
def test_build_request_no_auth_when_no_config(self, monkeypatch):
|
||
from specify_cli.authentication.http import build_request
|
||
self._set_config(monkeypatch, [])
|
||
req = build_request("https://github.com/org/repo")
|
||
assert "Authorization" not in req.headers
|
||
|
||
def test_build_request_extra_headers(self, monkeypatch):
|
||
from specify_cli.authentication.http import build_request
|
||
monkeypatch.setenv("GH_TOKEN", "my-token")
|
||
self._set_config(monkeypatch, [_github_entry()])
|
||
req = build_request("https://github.com/api", extra_headers={"Accept": "application/json"})
|
||
assert req.get_header("Accept") == "application/json"
|
||
assert req.get_header("Authorization") == "Bearer my-token"
|
||
|
||
def test_open_url_attaches_auth_for_matching_host(self, monkeypatch):
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli.authentication.http import open_url
|
||
monkeypatch.setenv("GH_TOKEN", "my-token")
|
||
self._set_config(monkeypatch, [_github_entry()])
|
||
captured = {}
|
||
mock_opener = MagicMock()
|
||
def fake_open(req, timeout=None):
|
||
captured["req"] = req
|
||
resp = MagicMock()
|
||
resp.__enter__ = lambda s: s
|
||
resp.__exit__ = MagicMock(return_value=False)
|
||
return resp
|
||
mock_opener.open.side_effect = fake_open
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
open_url("https://github.com/org/repo/catalog.json")
|
||
assert captured["req"].get_header("Authorization") == "Bearer my-token"
|
||
|
||
def test_open_url_no_auth_for_non_matching_host(self, monkeypatch):
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli.authentication.http import open_url
|
||
monkeypatch.setenv("GH_TOKEN", "my-token")
|
||
self._set_config(monkeypatch, [_github_entry()])
|
||
captured = {}
|
||
def fake_open(req, timeout=None):
|
||
captured["req"] = req
|
||
resp = MagicMock()
|
||
resp.__enter__ = lambda s: s
|
||
resp.__exit__ = MagicMock(return_value=False)
|
||
return resp
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = fake_open
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
open_url("https://example.com/file.json")
|
||
assert captured["req"].get_header("Authorization") is None
|
||
|
||
def test_open_url_no_auth_when_no_config(self, monkeypatch):
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli.authentication.http import open_url
|
||
self._set_config(monkeypatch, [])
|
||
captured = {}
|
||
def fake_open(req, timeout=None):
|
||
captured["req"] = req
|
||
resp = MagicMock()
|
||
resp.__enter__ = lambda s: s
|
||
resp.__exit__ = MagicMock(return_value=False)
|
||
return resp
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = fake_open
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
open_url("https://github.com/org/repo")
|
||
assert captured["req"].get_header("Authorization") is None
|
||
|
||
def test_open_url_falls_through_on_401(self, monkeypatch):
|
||
import urllib.error
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli.authentication.http import open_url
|
||
monkeypatch.setenv("GH_TOKEN", "bad-token")
|
||
self._set_config(monkeypatch, [_github_entry()])
|
||
call_count = 0
|
||
def fake_side_effect(req, timeout=None):
|
||
nonlocal call_count
|
||
call_count += 1
|
||
if call_count == 1:
|
||
raise urllib.error.HTTPError("url", 401, "Unauthorized", {}, None)
|
||
resp = MagicMock()
|
||
resp.__enter__ = lambda s: s
|
||
resp.__exit__ = MagicMock(return_value=False)
|
||
return resp
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = fake_side_effect
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
open_url("https://github.com/org/repo")
|
||
assert call_count == 2
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# open_url — negative tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestAuthenticatedHttpNegative:
|
||
def _set_config(self, monkeypatch, entries):
|
||
from specify_cli.authentication import http as _mod
|
||
monkeypatch.setattr(_mod, "_config_override", entries)
|
||
|
||
def test_500_raises_immediately(self, monkeypatch):
|
||
import urllib.error
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli.authentication.http import open_url
|
||
monkeypatch.setenv("GH_TOKEN", "tok")
|
||
self._set_config(monkeypatch, [_github_entry()])
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = urllib.error.HTTPError("url", 500, "ISE", {}, None)
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
with pytest.raises(urllib.error.HTTPError, match="500"):
|
||
open_url("https://github.com/org/repo")
|
||
|
||
def test_404_raises_immediately(self, monkeypatch):
|
||
import urllib.error
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli.authentication.http import open_url
|
||
monkeypatch.setenv("GH_TOKEN", "tok")
|
||
self._set_config(monkeypatch, [_github_entry()])
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = urllib.error.HTTPError("url", 404, "Not Found", {}, None)
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
with pytest.raises(urllib.error.HTTPError, match="404"):
|
||
open_url("https://github.com/org/repo")
|
||
|
||
def test_urlerror_propagates(self, monkeypatch):
|
||
import urllib.error
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli.authentication.http import open_url
|
||
self._set_config(monkeypatch, [])
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = urllib.error.URLError("refused")
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
with pytest.raises(urllib.error.URLError):
|
||
open_url("https://example.com/file")
|
||
|
||
def test_timeout_propagates(self, monkeypatch):
|
||
import socket
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli.authentication.http import open_url
|
||
self._set_config(monkeypatch, [])
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = socket.timeout("timed out")
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
with pytest.raises(socket.timeout):
|
||
open_url("https://example.com/file")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _load_config caching
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestLoadConfigCaching:
|
||
def test_config_cached_after_first_load(self, monkeypatch):
|
||
"""_load_config() should call load_auth_config only once per process."""
|
||
from unittest.mock import patch
|
||
from specify_cli.authentication import http as _mod
|
||
# Allow the real load path (no override)
|
||
monkeypatch.setattr(_mod, "_config_override", None)
|
||
monkeypatch.setattr(_mod, "_config_cache", None)
|
||
|
||
entry = _github_entry()
|
||
call_count = 0
|
||
|
||
def fake_load(path=None):
|
||
nonlocal call_count
|
||
call_count += 1
|
||
return [entry]
|
||
|
||
with patch.object(_mod, "load_auth_config", side_effect=fake_load):
|
||
_mod._load_config()
|
||
_mod._load_config()
|
||
_mod._load_config()
|
||
|
||
assert call_count == 1
|
||
|
||
def test_cache_bypassed_by_override(self, monkeypatch):
|
||
"""When _config_override is set, the cache is ignored entirely."""
|
||
from specify_cli.authentication import http as _mod
|
||
sentinel = [_github_entry()]
|
||
monkeypatch.setattr(_mod, "_config_override", sentinel)
|
||
monkeypatch.setattr(_mod, "_config_cache", None)
|
||
|
||
result = _mod._load_config()
|
||
assert result is sentinel
|
||
# Cache must not have been populated when override is active
|
||
assert _mod._config_cache is None
|
||
|
||
def test_failed_load_warns_once_and_caches_empty(self, monkeypatch):
|
||
"""A bad auth.json emits exactly one warning and subsequent calls use cache."""
|
||
from unittest.mock import patch
|
||
from specify_cli.authentication import http as _mod
|
||
import warnings as _warnings
|
||
monkeypatch.setattr(_mod, "_config_override", None)
|
||
monkeypatch.setattr(_mod, "_config_cache", None)
|
||
|
||
call_count = 0
|
||
|
||
def fail_load(path=None):
|
||
nonlocal call_count
|
||
call_count += 1
|
||
raise ValueError("bad config")
|
||
|
||
with patch.object(_mod, "load_auth_config", side_effect=fail_load):
|
||
with _warnings.catch_warnings(record=True) as w:
|
||
_warnings.simplefilter("always")
|
||
result1 = _mod._load_config()
|
||
result2 = _mod._load_config()
|
||
result3 = _mod._load_config()
|
||
|
||
user_warnings = [x for x in w if issubclass(x.category, UserWarning)]
|
||
assert len(user_warnings) == 1, "Expected exactly one warning"
|
||
# Loader called only once — subsequent calls used cache
|
||
assert call_count == 1
|
||
# All calls returned the cached empty list
|
||
assert result1 == result2 == result3 == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Redirect stripping
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRedirectStripping:
|
||
def test_redirect_within_hosts_preserves_auth(self):
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
from urllib.request import Request
|
||
import io
|
||
handler = _StripAuthOnRedirect(("github.com", "codeload.github.com"))
|
||
req = Request("https://github.com/org/repo", headers={"Authorization": "Bearer tok"})
|
||
new_req = handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||
"https://codeload.github.com/org/repo/zip")
|
||
assert new_req is not None
|
||
auth = new_req.get_header("Authorization") or new_req.unredirected_hdrs.get("Authorization")
|
||
assert auth == "Bearer tok"
|
||
|
||
def test_redirect_outside_hosts_strips_auth(self):
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
from urllib.request import Request
|
||
import io
|
||
handler = _StripAuthOnRedirect(("github.com",))
|
||
req = Request("https://github.com/org/repo", headers={"Authorization": "Bearer tok"})
|
||
new_req = handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||
"https://objects.githubusercontent.com/asset")
|
||
assert new_req is not None
|
||
assert new_req.headers.get("Authorization") is None
|
||
assert new_req.unredirected_hdrs.get("Authorization") is None
|
||
|
||
def test_https_to_http_same_host_redirect_rejected(self):
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
from urllib.request import Request
|
||
import io
|
||
import urllib.error
|
||
|
||
handler = _StripAuthOnRedirect(("github.com",))
|
||
req = Request("https://github.com/org/repo", headers={"Authorization": "Bearer tok"})
|
||
|
||
with pytest.raises(urllib.error.URLError, match="unsafe redirect"):
|
||
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||
"http://github.com/org/repo")
|
||
|
||
def test_redirect_validator_can_reject_before_following_redirect(self):
|
||
import urllib.error
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
from urllib.request import Request
|
||
import io
|
||
|
||
def reject_http(old_url, new_url):
|
||
if new_url.startswith("http://"):
|
||
raise urllib.error.URLError("scheme downgrade")
|
||
|
||
handler = _StripAuthOnRedirect(("github.com",), reject_http)
|
||
req = Request("https://github.com/org/repo", headers={"Authorization": "Bearer tok"})
|
||
|
||
with pytest.raises(urllib.error.URLError, match="scheme downgrade"):
|
||
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||
"http://github.com/org/repo")
|
||
|
||
def test_multi_hop_redirect_within_hosts_preserves_auth(self):
|
||
"""Auth survives a multi-hop redirect chain within allowed hosts."""
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
from urllib.request import Request
|
||
import io
|
||
hosts = ("github.com", "codeload.github.com", "objects-origin.githubusercontent.com")
|
||
handler = _StripAuthOnRedirect(hosts)
|
||
|
||
# First hop: github.com → codeload.github.com
|
||
req1 = Request("https://github.com/org/repo", headers={"Authorization": "Bearer tok"})
|
||
req2 = handler.redirect_request(req1, io.BytesIO(b""), 302, "Found", {},
|
||
"https://codeload.github.com/org/repo/zip")
|
||
assert req2 is not None
|
||
auth2 = req2.get_header("Authorization") or req2.unredirected_hdrs.get("Authorization")
|
||
assert auth2 == "Bearer tok"
|
||
|
||
# Second hop: codeload.github.com → objects-origin.githubusercontent.com
|
||
req3 = handler.redirect_request(req2, io.BytesIO(b""), 302, "Found", {},
|
||
"https://objects-origin.githubusercontent.com/asset")
|
||
assert req3 is not None
|
||
auth3 = req3.get_header("Authorization") or req3.unredirected_hdrs.get("Authorization")
|
||
assert auth3 == "Bearer tok"
|
||
|
||
def test_malformed_redirect_url_raises_urlerror_not_valueerror(self):
|
||
"""A redirect to a malformed URL (unterminated IPv6 bracket) surfaces
|
||
as URLError, which download paths already handle, rather than an
|
||
unhandled ValueError traceback."""
|
||
import urllib.error
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
from urllib.request import Request
|
||
import io
|
||
|
||
handler = _StripAuthOnRedirect(("github.com",))
|
||
req = Request("https://github.com/org/repo")
|
||
|
||
with pytest.raises(urllib.error.URLError):
|
||
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||
"https://[::1/asset")
|
||
|
||
def test_redirect_rejects_https_downgrade(self):
|
||
"""HTTPS downloads must not follow redirects to non-local HTTP URLs."""
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
from urllib.request import Request
|
||
import io
|
||
import urllib.error
|
||
handler = _StripAuthOnRedirect(("example.com",))
|
||
req = Request("https://example.com/archive.zip")
|
||
with pytest.raises(urllib.error.URLError, match="unsafe redirect"):
|
||
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||
"http://evil.example.com/archive.zip")
|
||
|
||
@pytest.mark.parametrize(
|
||
"target",
|
||
[
|
||
"http://127.0.0.1/internal",
|
||
"https://localhost/internal",
|
||
"https://localhost./internal",
|
||
"https://service.localhost/internal",
|
||
"https://service.localhost./internal",
|
||
"https://127.0.0.2/internal",
|
||
"https://127.1/internal",
|
||
"https://2130706433/internal",
|
||
"https://0x7f000001/internal",
|
||
"https://017700000001/internal",
|
||
"https://0177.0.0.1/internal",
|
||
"https://[::1]/internal",
|
||
"https://[::1%25lo0]/internal",
|
||
"https://[::ffff:127.0.0.1]/internal",
|
||
"https://127%2e0%2e0%2e1/internal",
|
||
"https://%31%32%37.0.0.1/internal",
|
||
"https://127%2E1/internal",
|
||
"https://local%68ost/internal",
|
||
"https://[::ffff:127%2e0.0.1]/internal",
|
||
"https://[::ffff:7f00%3a1]/internal",
|
||
"https://[::ffff%3a127.0.0.1]/internal",
|
||
"https://ℓocalhost/internal",
|
||
"https://localhost/internal",
|
||
"https://127。0。0。1/internal",
|
||
"https://0.0.0.0/internal",
|
||
"https://0/internal",
|
||
"https://00.00.00.00/internal",
|
||
"https://[::]/internal",
|
||
"https://[::ffff:0.0.0.0]/internal",
|
||
],
|
||
)
|
||
def test_redirect_rejects_remote_to_loopback(self, target):
|
||
"""A remote response must not redirect a download into loopback."""
|
||
import io
|
||
import urllib.error
|
||
from urllib.request import Request
|
||
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
|
||
handler = _StripAuthOnRedirect(())
|
||
request = Request("https://example.com/archive.zip")
|
||
|
||
with pytest.raises(urllib.error.URLError, match="unsafe redirect"):
|
||
handler.redirect_request(
|
||
request,
|
||
io.BytesIO(b""),
|
||
302,
|
||
"Found",
|
||
{},
|
||
target,
|
||
)
|
||
|
||
@pytest.mark.parametrize(
|
||
("source", "target"),
|
||
[
|
||
(
|
||
"http://localhost:8000/archive.zip",
|
||
"http://127.0.0.1:8001/archive.zip",
|
||
),
|
||
(
|
||
"http://127.0.0.2:8000/archive.zip",
|
||
"http://127.255.255.254:8001/archive.zip",
|
||
),
|
||
(
|
||
"https://[0:0:0:0:0:0:0:1]/archive.zip",
|
||
"http://[::1]:8001/archive.zip",
|
||
),
|
||
],
|
||
)
|
||
def test_redirect_allows_loopback_to_http_loopback(self, source, target):
|
||
"""Local development may continue redirecting between loopback URLs."""
|
||
import io
|
||
from urllib.request import Request
|
||
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
|
||
handler = _StripAuthOnRedirect(())
|
||
request = Request(source)
|
||
redirected = handler.redirect_request(
|
||
request,
|
||
io.BytesIO(b""),
|
||
302,
|
||
"Found",
|
||
{},
|
||
target,
|
||
)
|
||
|
||
assert redirected is not None
|
||
|
||
def test_multi_hop_remote_to_loopback_chain_is_rejected_at_first_hop(self):
|
||
import io
|
||
import urllib.error
|
||
from urllib.request import Request
|
||
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
|
||
handler = _StripAuthOnRedirect(())
|
||
request = Request("https://example.com/archive.zip")
|
||
|
||
with pytest.raises(urllib.error.URLError, match="unsafe redirect"):
|
||
handler.redirect_request(
|
||
request,
|
||
io.BytesIO(b""),
|
||
302,
|
||
"Found",
|
||
{},
|
||
"https://localhost:4443/hop",
|
||
)
|
||
|
||
@pytest.mark.parametrize(
|
||
"target",
|
||
[
|
||
"https://example.com:notaport/archive.zip",
|
||
"https://example.com:+443/archive.zip",
|
||
"https://example.com:65536/archive.zip",
|
||
],
|
||
)
|
||
def test_malformed_redirect_port_raises_urlerror(self, target):
|
||
import io
|
||
import urllib.error
|
||
from urllib.request import Request
|
||
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
|
||
handler = _StripAuthOnRedirect(())
|
||
request = Request("https://example.com/archive.zip")
|
||
|
||
with pytest.raises(urllib.error.URLError, match="malformed redirect URL"):
|
||
handler.redirect_request(
|
||
request,
|
||
io.BytesIO(b""),
|
||
302,
|
||
"Found",
|
||
{},
|
||
target,
|
||
)
|
||
|
||
def test_strict_redirect_error_describes_target_and_allowed_localhost(self):
|
||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||
from urllib.request import Request
|
||
import io
|
||
import urllib.error
|
||
|
||
handler = _StripAuthOnRedirect(("example.com",))
|
||
req = Request("https://example.com/archive.zip")
|
||
|
||
with pytest.raises(urllib.error.URLError) as exc_info:
|
||
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||
"http://evil.example.com/archive.zip")
|
||
|
||
error_message = str(exc_info.value)
|
||
assert "http://evil.example.com/archive.zip" in error_message
|
||
assert "localhost" in error_message
|
||
assert "127.0.0.1" in error_message
|
||
assert "::1" in error_message
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _fetch_latest_release_tag delegation
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFetchLatestReleaseTagDelegation:
|
||
def _set_config(self, monkeypatch, entries):
|
||
from specify_cli.authentication import http as _mod
|
||
monkeypatch.setattr(_mod, "_config_override", entries)
|
||
|
||
def _capture_request(self):
|
||
import json as _json
|
||
from unittest.mock import MagicMock
|
||
captured: dict = {}
|
||
def side_effect(req, timeout=None):
|
||
captured["request"] = req
|
||
body = _json.dumps({"tag_name": "v9.9.9"}).encode()
|
||
resp = MagicMock()
|
||
resp.read.side_effect = io.BytesIO(body).read
|
||
cm = MagicMock()
|
||
cm.__enter__.return_value = resp
|
||
cm.__exit__.return_value = False
|
||
return cm
|
||
return captured, side_effect
|
||
|
||
def test_gh_token_forwarded_when_configured(self, monkeypatch):
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli._version import _fetch_latest_release_tag
|
||
monkeypatch.setenv("GH_TOKEN", "forwarded-sentinel")
|
||
self._set_config(monkeypatch, [_github_entry()])
|
||
captured, side_effect = self._capture_request()
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = side_effect
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
_fetch_latest_release_tag()
|
||
assert captured["request"].get_header("Authorization") == "Bearer forwarded-sentinel"
|
||
|
||
def test_no_config_means_no_auth(self, monkeypatch):
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli._version import _fetch_latest_release_tag
|
||
self._set_config(monkeypatch, [])
|
||
captured, side_effect = self._capture_request()
|
||
# The unauthenticated path uses the strict redirect opener too.
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = side_effect
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
_fetch_latest_release_tag()
|
||
assert captured["request"].get_header("Authorization") is None
|
||
|
||
def test_accept_header_present(self, monkeypatch):
|
||
from unittest.mock import MagicMock, patch
|
||
from specify_cli._version import _fetch_latest_release_tag
|
||
self._set_config(monkeypatch, [])
|
||
captured, side_effect = self._capture_request()
|
||
mock_opener = MagicMock()
|
||
mock_opener.open.side_effect = side_effect
|
||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||
_fetch_latest_release_tag()
|
||
assert captured["request"].get_header("Accept") == "application/vnd.github+json"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# github_provider_hosts
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGithubProviderHosts:
|
||
"""Tests for github_provider_hosts() — the GHES host allowlist source."""
|
||
|
||
def _set_config(self, monkeypatch, entries):
|
||
from specify_cli.authentication import http as _auth_http
|
||
monkeypatch.setattr(_auth_http, "_config_override", entries)
|
||
|
||
def test_returns_hosts_from_github_entries(self, monkeypatch):
|
||
from specify_cli.authentication.http import github_provider_hosts
|
||
self._set_config(monkeypatch, [
|
||
AuthConfigEntry(hosts=("ghes.example", "raw.ghes.example"),
|
||
provider="github", auth="bearer", token="t"),
|
||
])
|
||
assert github_provider_hosts() == ("ghes.example", "raw.ghes.example")
|
||
|
||
def test_empty_when_no_config(self, monkeypatch):
|
||
from specify_cli.authentication.http import github_provider_hosts
|
||
self._set_config(monkeypatch, [])
|
||
assert github_provider_hosts() == ()
|
||
|
||
def test_ignores_non_github_providers(self, monkeypatch):
|
||
from specify_cli.authentication.http import github_provider_hosts
|
||
self._set_config(monkeypatch, [
|
||
AuthConfigEntry(hosts=("dev.azure.com",), provider="azure-devops",
|
||
auth="basic-pat", token="t"),
|
||
])
|
||
assert github_provider_hosts() == ()
|
||
|
||
def test_unions_multiple_github_entries(self, monkeypatch):
|
||
from specify_cli.authentication.http import github_provider_hosts
|
||
self._set_config(monkeypatch, [
|
||
AuthConfigEntry(hosts=("ghes.example",), provider="github", auth="bearer", token="t"),
|
||
AuthConfigEntry(hosts=("github.com",), provider="github", auth="bearer", token="t"),
|
||
])
|
||
assert github_provider_hosts() == ("ghes.example", "github.com")
|