# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import base64
import importlib.util
import sys
from pathlib import Path
from typing import cast
import pytest
MODULE_PATH = Path(__file__).with_name("github_actions_policy.py")
SPEC = importlib.util.spec_from_file_location("github_actions_policy", MODULE_PATH)
assert SPEC is not None
assert SPEC.loader is not None
policy = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = policy
SPEC.loader.exec_module(policy)
def test_discover_references_preserves_locations_and_supported_forms() -> None:
source = """
jobs:
build:
uses: owner/reusable/.github/workflows/build.yml@v1
steps:
- uses: actions/checkout@0123456789012345678901234567890123456789 # pinned
- uses: './.github/actions/local'
- uses: "docker://alpine:3.22"
- {"uses": owner/inline@sha}
"""
assert policy.discover_references(source, ".github/workflows/ci.yml") == [
policy.Reference(
"owner/reusable/.github/workflows/build.yml@v1",
".github/workflows/ci.yml",
4,
"job",
),
policy.Reference(
"actions/checkout@0123456789012345678901234567890123456789",
".github/workflows/ci.yml",
6,
"step",
),
policy.Reference(
"./.github/actions/local", ".github/workflows/ci.yml", 7, "step"
),
policy.Reference("docker://alpine:3.22", ".github/workflows/ci.yml", 8, "step"),
policy.Reference("owner/inline@sha", ".github/workflows/ci.yml", 9, "step"),
]
@pytest.mark.parametrize(
"source",
[
"jobs:\n build:\n steps:\n - uses: |\n owner/repo@v1\n",
"jobs:\n build:\n steps:\n - uses: ${{ matrix.action }}\n",
"jobs:\n build:\n steps:\n - uses: [owner/repo@v1]\n",
],
)
def test_discover_references_rejects_ambiguous_uses_syntax(source: str) -> None:
with pytest.raises(policy.EvaluationError, match="uses"):
policy.discover_references(source, ".github/workflows/ci.yml")
def test_discover_references_finds_quoted_uses_key() -> None:
source = 'jobs:\n build:\n steps:\n - "uses": attacker/action@v1\n'
assert policy.discover_references(source, ".github/workflows/ci.yml") == [
policy.Reference("attacker/action@v1", ".github/workflows/ci.yml", 4, "step")
]
def test_added_references_compares_occurrence_counts() -> None:
base = [
policy.Reference("owner/action@old", "ci.yml", 2, "step"),
policy.Reference("owner/action@same", "ci.yml", 3, "step"),
]
head = [
policy.Reference("owner/action@same", "ci.yml", 4, "step"),
policy.Reference("owner/action@new", "ci.yml", 5, "step"),
policy.Reference("owner/action@new", "ci.yml", 6, "step"),
]
assert policy.added_references(base, head) == head[1:]
def test_added_references_detects_step_to_job_context_change() -> None:
base_source = """
jobs:
build:
steps:
- uses: actions/checkout@0123456789012345678901234567890123456789
"""
head_source = """
jobs:
build:
uses: actions/checkout@0123456789012345678901234567890123456789
"""
base = policy.discover_references(base_source, ".github/workflows/ci.yml")
head = policy.discover_references(head_source, ".github/workflows/ci.yml")
assert policy.added_references(base, head) == head
def test_discover_references_ignores_uses_text_inside_block_scalar() -> None:
source = """
jobs:
build:
steps:
- run: |
echo "uses: attacker/action@main"
- uses: owner/action@sha
"""
assert policy.discover_references(source, ".github/workflows/ci.yml") == [
policy.Reference("owner/action@sha", ".github/workflows/ci.yml", 7, "step")
]
def test_discover_references_ignores_non_executable_uses_keys() -> None:
source = """
env:
uses: not/an/action
jobs:
build:
steps:
- uses: owner/action@sha
with:
uses: arbitrary-input
"""
assert policy.discover_references(source, ".github/workflows/ci.yml") == [
policy.Reference("owner/action@sha", ".github/workflows/ci.yml", 7, "step")
]
def test_discover_references_finds_composite_action_steps_only() -> None:
source = """
inputs:
uses:
default: arbitrary-input
runs:
using: composite
steps:
- uses: owner/action@sha
with:
uses: arbitrary-input
"""
assert policy.discover_references(source, ".github/actions/test/action.yml") == [
policy.Reference(
"owner/action@sha", ".github/actions/test/action.yml", 8, "step"
)
]
def selected_policy(**overrides: object) -> object:
values: dict[str, object] = {
"enabled": True,
"allowed_actions": "selected",
"sha_pinning_required": False,
"github_owned_allowed": False,
"verified_allowed": False,
"patterns_allowed": (),
}
values.update(overrides)
return policy.ActionsPolicy(**values)
def test_job_action_reference_is_rejected() -> None:
source = """
jobs:
build:
uses: actions/checkout@0123456789012345678901234567890123456789
"""
[reference] = policy.discover_references(source, ".github/workflows/ci.yml")
decision = policy.evaluate_reference(
reference,
selected_policy(sha_pinning_required=True, github_owned_allowed=True),
repository_owner="NVIDIA",
is_verified_marketplace_action=lambda _owner, _repo: False,
)
assert not decision.allowed
assert "job-level uses must reference a reusable workflow" in decision.reason
def test_step_workflow_path_does_not_bypass_sha_pinning() -> None:
source = """
jobs:
build:
steps:
- uses: NVIDIA/tools/.github/workflows/helper@v1
"""
[reference] = policy.discover_references(source, ".github/workflows/ci.yml")
decision = policy.evaluate_reference(
reference,
selected_policy(sha_pinning_required=True),
repository_owner="NVIDIA",
is_verified_marketplace_action=lambda _owner, _repo: False,
)
assert not decision.allowed
assert "full-length commit SHA" in decision.reason
@pytest.mark.parametrize(
"value",
["./.github/actions/local", "$/shared/action", "docker://alpine:3.22"],
)
def test_local_and_container_references_do_not_need_external_policy(value: str) -> None:
decision = policy.evaluate_reference(
policy.Reference(value, "ci.yml", 1, "step"),
selected_policy(),
repository_owner="NVIDIA",
is_verified_marketplace_action=lambda _owner, _repo: False,
)
assert decision.allowed
@pytest.mark.parametrize(
("value", "policy_overrides", "reason"),
[
("NVIDIA/internal-action@v1", {}, "repository owner"),
(
"actions/checkout@v4",
{"github_owned_allowed": True},
"GitHub-owned",
),
(
"github/codeql-action/init@v3",
{"github_owned_allowed": True},
"GitHub-owned",
),
(
"oras-project/setup-oras@approved",
{"patterns_allowed": ("oras-project/setup-oras@approved",)},
"explicit policy pattern",
),
(
"softprops/action-gh-release@sha",
{"patterns_allowed": ("softprops/action-gh-release@*",)},
"explicit policy pattern",
),
],
)
def test_selected_policy_allows_supported_categories(
value: str, policy_overrides: dict[str, object], reason: str
) -> None:
decision = policy.evaluate_reference(
policy.Reference(value, "ci.yml", 1, "step"),
selected_policy(**policy_overrides),
repository_owner="NVIDIA",
is_verified_marketplace_action=lambda _owner, _repo: False,
)
assert decision.allowed
assert reason in decision.reason
def test_verified_marketplace_applies_to_actions_but_not_reusable_workflows() -> None:
live_policy = selected_policy(verified_allowed=True)
def lookup(owner: str, repo: str) -> bool:
return (owner, repo) == ("astral-sh", "setup-uv")
action = policy.evaluate_reference(
policy.Reference("astral-sh/setup-uv@sha", "ci.yml", 1, "step"),
live_policy,
repository_owner="NVIDIA",
is_verified_marketplace_action=lookup,
)
workflow = policy.evaluate_reference(
policy.Reference(
"astral-sh/setup-uv/.github/workflows/test.yml@main",
"ci.yml",
2,
"job",
),
live_policy,
repository_owner="NVIDIA",
is_verified_marketplace_action=lookup,
)
assert action.allowed
assert "Marketplace-verified" in action.reason
assert not workflow.allowed
def test_sha_pinning_applies_to_external_actions_not_reusable_workflows() -> None:
live_policy = selected_policy(
sha_pinning_required=True,
patterns_allowed=("owner/repo@*", "owner/repo/.github/workflows/ci.yml@*"),
)
unpinned = policy.evaluate_reference(
policy.Reference("owner/repo@v1", "ci.yml", 1, "step"),
live_policy,
repository_owner="NVIDIA",
is_verified_marketplace_action=lambda _owner, _repo: False,
)
pinned = policy.evaluate_reference(
policy.Reference(
"owner/repo@0123456789012345678901234567890123456789",
"ci.yml",
2,
"step",
),
live_policy,
repository_owner="NVIDIA",
is_verified_marketplace_action=lambda _owner, _repo: False,
)
reusable = policy.evaluate_reference(
policy.Reference(
"owner/repo/.github/workflows/ci.yml@main", "ci.yml", 3, "job"
),
live_policy,
repository_owner="NVIDIA",
is_verified_marketplace_action=lambda _owner, _repo: False,
)
assert not unpinned.allowed
assert "full-length commit SHA" in unpinned.reason
assert pinned.allowed
assert reusable.allowed
def test_negative_pattern_after_allow_pattern_wins() -> None:
live_policy = selected_policy(
patterns_allowed=("owner/*", "!owner/blocked@*"),
)
decision = policy.evaluate_reference(
policy.Reference("owner/blocked@sha", "ci.yml", 1, "step"),
live_policy,
repository_owner="NVIDIA",
is_verified_marketplace_action=lambda _owner, _repo: False,
)
assert not decision.allowed
assert "blocked by policy pattern" in decision.reason
def test_policy_pattern_owner_and_repository_are_case_insensitive() -> None:
live_policy = selected_policy(
patterns_allowed=("swatinem/rust-cache@sha",),
)
decision = policy.evaluate_reference(
policy.Reference("Swatinem/rust-cache@sha", "ci.yml", 1, "step"),
live_policy,
repository_owner="NVIDIA",
is_verified_marketplace_action=lambda _owner, _repo: False,
)
assert decision.allowed
def test_enterprise_owned_organization_is_allowed() -> None:
decision = policy.evaluate_reference(
policy.Reference("sibling-org/action@sha", "ci.yml", 1, "step"),
selected_policy(),
repository_owner="NVIDIA",
enterprise_organizations=frozenset({"nvidia", "sibling-org"}),
is_verified_marketplace_action=lambda _owner, _repo: False,
)
assert decision.allowed
assert "enterprise" in decision.reason
class FakeWeb:
def __init__(self, responses: dict[str, str | Exception]) -> None:
self.responses = responses
self.calls: list[str] = []
def get_text(self, url: str) -> str:
self.calls.append(url)
response = self.responses[url]
if isinstance(response, Exception):
raise response
return response
def test_marketplace_verification_is_bound_to_exact_repository() -> None:
repository_url = "https://github.com/example/action"
listing_url = "https://github.com/marketplace/actions/example"
web = FakeWeb(
{
repository_url: 'Marketplace',
listing_url: (
'{"isVerifiedOwner":true,"externalUsesPathPrefix":"different/action@"}'
),
}
)
client = policy.MarketplaceClient(web)
assert not client.is_verified_action("example", "action")
def test_marketplace_verification_accepts_exact_repository() -> None:
repository_url = "https://github.com/docker/setup-buildx-action"
listing_url = "https://github.com/marketplace/actions/docker-setup-buildx"
web = FakeWeb(
{
repository_url: (
'Marketplace'
),
listing_url: (
'{"isVerifiedOwner":true,'
'"externalUsesPathPrefix":"docker/setup-buildx-action@"}'
),
}
)
client = policy.MarketplaceClient(web)
assert client.is_verified_action("docker", "setup-buildx-action")
assert client.is_verified_action("docker", "setup-buildx-action")
assert web.calls.count(repository_url) == 1
def test_marketplace_repository_identity_is_case_insensitive() -> None:
repository_url = "https://github.com/Docker/Setup-Buildx-Action"
listing_url = "https://github.com/marketplace/actions/docker-setup-buildx"
web = FakeWeb(
{
repository_url: (
'Marketplace'
),
listing_url: (
'{"isVerifiedOwner":true,'
'"externalUsesPathPrefix":"docker/setup-buildx-action@"}'
),
}
)
assert policy.MarketplaceClient(web).is_verified_action(
"Docker", "Setup-Buildx-Action"
)
def test_marketplace_ambiguous_metadata_fails_closed() -> None:
repository_url = "https://github.com/example/action"
listing_url = "https://github.com/marketplace/actions/example"
web = FakeWeb(
{
repository_url: 'Marketplace',
listing_url: '{"externalUsesPathPrefix":"example/action@"}',
}
)
with pytest.raises(policy.EvaluationError, match="verification metadata"):
policy.MarketplaceClient(web).is_verified_action("example", "action")
def test_policy_client_reads_selected_policy_without_copying_patterns() -> None:
permissions_url = (
"https://api.github.com/repos/NVIDIA/OpenShell/actions/permissions"
)
selected_url = f"{permissions_url}/selected-actions"
api = FakeWeb(
{
permissions_url: (
'{"enabled":true,"allowed_actions":"selected",'
'"sha_pinning_required":false}'
),
selected_url: (
'{"github_owned_allowed":true,"verified_allowed":true,'
'"patterns_allowed":["owner/action@sha"]}'
),
}
)
live_policy = policy.PolicyClient(api).get_policy("NVIDIA/OpenShell")
assert live_policy == selected_policy(
github_owned_allowed=True,
verified_allowed=True,
patterns_allowed=("owner/action@sha",),
)
def test_policy_client_rejects_invalid_or_disabled_policy() -> None:
permissions_url = (
"https://api.github.com/repos/NVIDIA/OpenShell/actions/permissions"
)
with pytest.raises(policy.EvaluationError, match="disabled"):
policy.PolicyClient(
FakeWeb(
{
permissions_url: (
'{"enabled":false,"allowed_actions":"selected",'
'"sha_pinning_required":false}'
)
}
)
).get_policy("NVIDIA/OpenShell")
class FakeGitHub:
def __init__(self, responses: dict[str, object]) -> None:
self.responses = responses
self.posts: list[tuple[str, dict[str, object]]] = []
def get_json(self, url: str) -> object:
response = self.responses[url]
if isinstance(response, Exception):
raise response
return response
def post_json(self, url: str, payload: object) -> object:
assert isinstance(payload, dict)
self.posts.append((url, cast("dict[str, object]", payload)))
return {"ok": True}
class FakeGraphQL:
def __init__(self, responses: list[object]) -> None:
self.responses = responses
def post_json(self, _url: str, _payload: object) -> object:
return self.responses.pop(0)
def test_enterprise_client_resolves_only_repository_owner_enterprise() -> None:
client = FakeGraphQL(
[
{
"data": {
"viewer": {
"enterprises": {
"nodes": [{"id": "enterprise-1"}],
"pageInfo": {"hasNextPage": False, "endCursor": None},
}
}
}
},
{
"data": {
"node": {
"organizations": {
"nodes": [{"login": "NVIDIA"}, {"login": "Sibling-Org"}],
"pageInfo": {"hasNextPage": False, "endCursor": None},
}
}
}
},
]
)
assert policy.EnterpriseClient(client).organizations_for("nvidia") == frozenset(
{"nvidia", "sibling-org"}
)
def test_source_client_decodes_wrapped_base64() -> None:
content = (
"jobs:\n test:\n uses: owner/workflows/.github/workflows/ci.yml@main\n"
)
encoded = base64.encodebytes(content.encode()).decode()
url = (
"https://api.github.com/repos/fork/OpenShell/contents/"
".github/workflows/ci.yml?ref=head"
)
client = FakeGitHub(
{url: {"encoding": "base64", "content": encoded, "size": len(content)}}
)
assert (
policy.GitHubSourceClient(client).content(
"fork/OpenShell", ".github/workflows/ci.yml", "head"
)
== content
)
def test_source_client_fails_closed_when_expected_content_is_missing() -> None:
url = (
"https://api.github.com/repos/fork/OpenShell/contents/"
".github/workflows/ci.yml?ref=head"
)
client = FakeGitHub({url: policy.HttpError(404, url, "Not Found")})
with pytest.raises(policy.HttpError, match="404"):
policy.GitHubSourceClient(client).content(
"fork/OpenShell", ".github/workflows/ci.yml", "head"
)
def test_source_client_fails_closed_on_incomplete_pull_file_list() -> None:
repository = "NVIDIA/OpenShell"
url = f"https://api.github.com/repos/{repository}/pulls/3624/files?per_page=100&page=1"
client = FakeGitHub({url: []})
with pytest.raises(policy.EvaluationError, match="incomplete"):
policy.GitHubSourceClient(client).pull_files(repository, 3624, 1)
def test_workflow_run_binds_pull_request_to_current_head() -> None:
head_sha = "b" * 40
base_sha = "a" * 40
repository = "NVIDIA/OpenShell"
pull_url = f"https://api.github.com/repos/{repository}/pulls/3624"
files_url = f"{pull_url}/files?per_page=100&page=1"
client = FakeGitHub(
{
pull_url: {
"number": 3624,
"changed_files": 0,
"base": {"sha": base_sha, "repo": {"full_name": repository}},
"head": {"sha": head_sha, "repo": {"full_name": "fork/OpenShell"}},
},
files_url: [],
}
)
event = {
"repository": {"full_name": repository},
"workflow_run": {
"event": "pull_request",
"conclusion": "success",
"head_sha": head_sha,
"path": ".github/workflows/workflow-policy-request.yml",
"pull_requests": [{"number": 3624}],
},
}
assert policy.run_workflow_run(event, client, "https://example.test/run") == 0
assert [payload["state"] for _, payload in client.posts] == [
"pending",
"success",
]
def test_workflow_run_skips_stale_pull_request_head() -> None:
run_sha = "a" * 40
current_sha = "b" * 40
repository = "NVIDIA/OpenShell"
pull_url = f"https://api.github.com/repos/{repository}/pulls/3624"
client = FakeGitHub(
{
pull_url: {
"number": 3624,
"changed_files": 0,
"base": {"sha": "c" * 40, "repo": {"full_name": repository}},
"head": {
"sha": current_sha,
"repo": {"full_name": "fork/OpenShell"},
},
}
}
)
event = {
"repository": {"full_name": repository},
"workflow_run": {
"event": "pull_request",
"conclusion": "success",
"head_sha": run_sha,
"path": ".github/workflows/workflow-policy-request.yml",
"pull_requests": [{"number": 3624}],
},
}
assert policy.run_workflow_run(event, client, "https://example.test/run") == 0
assert client.posts == []
def test_merge_group_scans_complete_candidate_tree() -> None:
merge_sha = "a" * 40
repository = "NVIDIA/OpenShell"
tree_url = (
f"https://api.github.com/repos/{repository}/git/trees/{merge_sha}?recursive=1"
)
client = FakeGitHub({tree_url: {"truncated": False, "tree": []}})
assert (
policy.run_merge_group(
repository=repository,
head_sha=merge_sha,
github_client=client,
target_url="https://example.test/run",
)
== 0
)
assert [payload["state"] for _, payload in client.posts] == [
"pending",
"success",
]
def test_merge_group_fails_closed_on_truncated_candidate_tree() -> None:
merge_sha = "a" * 40
repository = "NVIDIA/OpenShell"
tree_url = (
f"https://api.github.com/repos/{repository}/git/trees/{merge_sha}?recursive=1"
)
client = FakeGitHub({tree_url: {"truncated": True, "tree": []}})
assert (
policy.run_merge_group(
repository=repository,
head_sha=merge_sha,
github_client=client,
target_url="https://example.test/run",
)
== 1
)
assert [payload["state"] for _, payload in client.posts] == [
"pending",
"failure",
]