mirror of
https://github.com/NVIDIA/OpenShell.git
synced 2026-10-12 04:39:51 +08:00
996 lines
38 KiB
Python
996 lines
38 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Validate changed GitHub Actions references against the live repository policy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import fnmatch
|
|
import html
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from collections import Counter
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Literal, Protocol, cast
|
|
|
|
import yaml
|
|
from yaml.nodes import MappingNode, Node, ScalarNode, SequenceNode
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Callable
|
|
|
|
API_ROOT = "https://api.github.com"
|
|
MARKETPLACE_ROOT = "https://github.com"
|
|
STATUS_CONTEXT = "OpenShell / Workflow Policy"
|
|
MAX_DEFINITION_BYTES = 1024 * 1024
|
|
MAX_DEFINITION_COUNT = 1000
|
|
MAX_TOTAL_DEFINITION_BYTES = 10 * 1024 * 1024
|
|
FULL_SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$")
|
|
MARKETPLACE_LINK_RE = re.compile(
|
|
r'href=["\']/marketplace/actions/([A-Za-z0-9_-]+)["\']'
|
|
)
|
|
|
|
|
|
class EvaluationError(RuntimeError):
|
|
"""A policy input could not be evaluated safely."""
|
|
|
|
|
|
class HttpError(EvaluationError):
|
|
def __init__(self, status: int, url: str, message: str) -> None:
|
|
super().__init__(f"GitHub request failed ({status}) for {url}: {message}")
|
|
self.status = status
|
|
self.url = url
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Reference:
|
|
value: str
|
|
path: str
|
|
line: int
|
|
context: Literal["job", "step"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ActionsPolicy:
|
|
enabled: bool
|
|
allowed_actions: str
|
|
sha_pinning_required: bool
|
|
github_owned_allowed: bool = False
|
|
verified_allowed: bool = False
|
|
patterns_allowed: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Decision:
|
|
allowed: bool
|
|
reason: str
|
|
|
|
|
|
class TextClient(Protocol):
|
|
def get_text(self, url: str) -> str: ...
|
|
|
|
|
|
class JsonClient(Protocol):
|
|
def post_json(self, url: str, payload: object) -> object: ...
|
|
|
|
|
|
def _parse_uses_value(value: str, path: str, line: int) -> str:
|
|
if not value:
|
|
raise EvaluationError(f"empty uses value at {path}:{line}")
|
|
if "${{" in value:
|
|
raise EvaluationError(
|
|
f"dynamic uses value cannot be policy-checked at {path}:{line}"
|
|
)
|
|
if any(character.isspace() for character in value):
|
|
raise EvaluationError(f"invalid scalar uses value at {path}:{line}")
|
|
return value
|
|
|
|
|
|
def discover_references(source: str, path: str) -> list[Reference]:
|
|
try:
|
|
document = yaml.compose(source, Loader=yaml.SafeLoader)
|
|
except yaml.YAMLError as exc:
|
|
raise EvaluationError(f"invalid YAML in {path}: {exc}") from exc
|
|
if document is None:
|
|
return []
|
|
|
|
if not isinstance(document, MappingNode):
|
|
raise EvaluationError(f"action definition must be a mapping in {path}")
|
|
|
|
def values_for(mapping: MappingNode, name: str) -> list[Node]:
|
|
return [
|
|
value
|
|
for key, value in mapping.value
|
|
if isinstance(key, ScalarNode) and key.value == name
|
|
]
|
|
|
|
references: list[Reference] = []
|
|
|
|
def collect_uses(mapping: MappingNode, context: Literal["job", "step"]) -> None:
|
|
for value in values_for(mapping, "uses"):
|
|
line = value.start_mark.line + 1
|
|
if (
|
|
not isinstance(value, ScalarNode)
|
|
or value.tag != "tag:yaml.org,2002:str"
|
|
):
|
|
raise EvaluationError(f"uses value must be a string at {path}:{line}")
|
|
references.append(
|
|
Reference(
|
|
_parse_uses_value(value.value, path, line), path, line, context
|
|
)
|
|
)
|
|
|
|
if re.fullmatch(r"\.github/workflows/[^/]+\.ya?ml", path, re.IGNORECASE):
|
|
for jobs_value in values_for(document, "jobs"):
|
|
if not isinstance(jobs_value, MappingNode):
|
|
continue
|
|
for _, job_value in jobs_value.value:
|
|
if not isinstance(job_value, MappingNode):
|
|
continue
|
|
collect_uses(job_value, "job")
|
|
for steps_value in values_for(job_value, "steps"):
|
|
if isinstance(steps_value, SequenceNode):
|
|
for step_value in steps_value.value:
|
|
if isinstance(step_value, MappingNode):
|
|
collect_uses(step_value, "step")
|
|
else:
|
|
for runs_value in values_for(document, "runs"):
|
|
if not isinstance(runs_value, MappingNode):
|
|
continue
|
|
for steps_value in values_for(runs_value, "steps"):
|
|
if isinstance(steps_value, SequenceNode):
|
|
for step_value in steps_value.value:
|
|
if isinstance(step_value, MappingNode):
|
|
collect_uses(step_value, "step")
|
|
return references
|
|
|
|
|
|
def added_references(base: list[Reference], head: list[Reference]) -> list[Reference]:
|
|
remaining = Counter((reference.value, reference.context) for reference in base)
|
|
added: list[Reference] = []
|
|
for reference in head:
|
|
identity = (reference.value, reference.context)
|
|
if remaining[identity] > 0:
|
|
remaining[identity] -= 1
|
|
else:
|
|
added.append(reference)
|
|
return added
|
|
|
|
|
|
def is_action_definition(path: str) -> bool:
|
|
lowered = path.lower()
|
|
if re.fullmatch(r"\.github/workflows/[^/]+\.ya?ml", lowered):
|
|
return True
|
|
return bool(re.fullmatch(r"\.github/actions/.+/action\.ya?ml", lowered))
|
|
|
|
|
|
def _split_external_reference(value: str) -> tuple[str, str, str, bool]:
|
|
if "@" not in value:
|
|
raise EvaluationError(
|
|
f"external reference {value!r} has no @ revision and cannot be evaluated"
|
|
)
|
|
target, revision = value.rsplit("@", 1)
|
|
parts = target.split("/")
|
|
if len(parts) < 2 or not parts[0] or not parts[1] or not revision:
|
|
raise EvaluationError(f"invalid external reference {value!r}")
|
|
reusable = (
|
|
len(parts) == 5
|
|
and parts[2:4] == [".github", "workflows"]
|
|
and re.fullmatch(r"[^/]+\.ya?ml", parts[4], re.IGNORECASE) is not None
|
|
)
|
|
return parts[0], parts[1], revision, reusable
|
|
|
|
|
|
def _pattern_decision(value: str, patterns: tuple[str, ...]) -> bool | None:
|
|
def normalize_identity(candidate: str) -> str:
|
|
target, separator, revision = candidate.rpartition("@")
|
|
parts = target.split("/")
|
|
if len(parts) >= 2:
|
|
parts[0] = parts[0].casefold()
|
|
parts[1] = parts[1].casefold()
|
|
normalized = "/".join(parts)
|
|
return f"{normalized}{separator}{revision}" if separator else normalized
|
|
|
|
normalized_value = normalize_identity(value)
|
|
decision: bool | None = None
|
|
for raw_pattern in patterns:
|
|
deny = raw_pattern.startswith("!")
|
|
pattern = raw_pattern[1:] if deny else raw_pattern
|
|
if pattern and fnmatch.fnmatchcase(
|
|
normalized_value, normalize_identity(pattern)
|
|
):
|
|
decision = not deny
|
|
return decision
|
|
|
|
|
|
def evaluate_reference(
|
|
reference: Reference,
|
|
policy: ActionsPolicy,
|
|
*,
|
|
repository_owner: str,
|
|
enterprise_organizations: frozenset[str] = frozenset(),
|
|
is_verified_marketplace_action: Callable[[str, str], bool],
|
|
) -> Decision:
|
|
value = reference.value
|
|
if reference.context == "job":
|
|
if re.fullmatch(r"\./\.github/workflows/[^/]+\.ya?ml", value, re.IGNORECASE):
|
|
return Decision(True, "local reusable workflow")
|
|
if value.startswith(("./", "$/", "docker://")):
|
|
return Decision(False, "job-level uses must reference a reusable workflow")
|
|
elif value.startswith(("./", "$/", "docker://")):
|
|
return Decision(True, "local or container reference")
|
|
if not policy.enabled:
|
|
return Decision(False, "GitHub Actions is disabled for the repository")
|
|
|
|
owner, repository, revision, reusable = _split_external_reference(value)
|
|
if reference.context == "job" and not reusable:
|
|
return Decision(False, "job-level uses must reference a reusable workflow")
|
|
reusable = reference.context == "job"
|
|
if (
|
|
policy.sha_pinning_required
|
|
and not reusable
|
|
and not FULL_SHA_RE.fullmatch(revision)
|
|
):
|
|
return Decision(False, "external actions must use a full-length commit SHA")
|
|
|
|
if policy.allowed_actions == "all":
|
|
return Decision(True, "repository policy allows all actions")
|
|
if policy.allowed_actions == "local_only":
|
|
return Decision(False, "repository policy allows only local actions")
|
|
if policy.allowed_actions != "selected":
|
|
raise EvaluationError(
|
|
f"unsupported allowed_actions value {policy.allowed_actions!r}"
|
|
)
|
|
|
|
pattern_decision = _pattern_decision(value, policy.patterns_allowed)
|
|
if pattern_decision is False:
|
|
return Decision(False, "reference is blocked by policy pattern")
|
|
if pattern_decision is True:
|
|
return Decision(True, "reference matches an explicit policy pattern")
|
|
|
|
if owner.casefold() == repository_owner.casefold():
|
|
return Decision(True, "reference belongs to the repository owner")
|
|
if owner.casefold() in enterprise_organizations:
|
|
return Decision(True, "reference belongs to the repository owner's enterprise")
|
|
if policy.github_owned_allowed and owner.casefold() in {"actions", "github"}:
|
|
return Decision(True, "reference is GitHub-owned")
|
|
if (
|
|
policy.verified_allowed
|
|
and not reusable
|
|
and is_verified_marketplace_action(owner, repository)
|
|
):
|
|
return Decision(True, "action belongs to a Marketplace-verified creator")
|
|
return Decision(
|
|
False, "reference is not allowed by the effective repository policy"
|
|
)
|
|
|
|
|
|
class HttpClient:
|
|
def __init__(self, token: str = "") -> None:
|
|
self.token = token
|
|
|
|
def _request(
|
|
self, url: str, *, method: str = "GET", payload: object | None = None
|
|
) -> str:
|
|
accept = (
|
|
"application/vnd.github+json"
|
|
if urllib.parse.urlsplit(url).hostname == "api.github.com"
|
|
else "text/html,application/xhtml+xml"
|
|
)
|
|
headers = {
|
|
"Accept": accept,
|
|
"User-Agent": "OpenShell-workflow-policy-check",
|
|
"X-GitHub-Api-Version": "2022-11-28",
|
|
}
|
|
if self.token:
|
|
headers["Authorization"] = f"Bearer {self.token}"
|
|
data = None
|
|
if payload is not None:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
request = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
body = response.read(MAX_DEFINITION_BYTES + 1)
|
|
except urllib.error.HTTPError as exc:
|
|
detail = exc.reason or "HTTP error"
|
|
raise HttpError(exc.code, url, str(detail)) from exc
|
|
except (urllib.error.URLError, TimeoutError) as exc:
|
|
raise EvaluationError(f"GitHub request failed for {url}: {exc}") from exc
|
|
if len(body) > MAX_DEFINITION_BYTES:
|
|
raise EvaluationError(f"response from {url} exceeds size limit")
|
|
return body.decode("utf-8")
|
|
|
|
def get_text(self, url: str) -> str:
|
|
return self._request(url)
|
|
|
|
def get_json(self, url: str) -> object:
|
|
try:
|
|
return json.loads(self.get_text(url))
|
|
except json.JSONDecodeError as exc:
|
|
raise EvaluationError(f"GitHub returned invalid JSON for {url}") from exc
|
|
|
|
def post_json(self, url: str, payload: object) -> object:
|
|
try:
|
|
return json.loads(self._request(url, method="POST", payload=payload))
|
|
except json.JSONDecodeError as exc:
|
|
raise EvaluationError(f"GitHub returned invalid JSON for {url}") from exc
|
|
|
|
|
|
def _load_mapping(client: TextClient, url: str) -> dict[str, object]:
|
|
try:
|
|
value = json.loads(client.get_text(url))
|
|
except json.JSONDecodeError as exc:
|
|
raise EvaluationError(f"GitHub returned invalid policy JSON for {url}") from exc
|
|
if not isinstance(value, dict):
|
|
raise EvaluationError(f"GitHub returned an invalid policy object for {url}")
|
|
return value
|
|
|
|
|
|
class PolicyClient:
|
|
def __init__(self, client: TextClient) -> None:
|
|
self.client = client
|
|
|
|
def get_policy(self, repository: str) -> ActionsPolicy:
|
|
permissions_url = f"{API_ROOT}/repos/{repository}/actions/permissions"
|
|
permissions = _load_mapping(self.client, permissions_url)
|
|
enabled = permissions.get("enabled")
|
|
allowed_actions = permissions.get("allowed_actions")
|
|
sha_pinning = permissions.get("sha_pinning_required", False)
|
|
if enabled is not True:
|
|
raise EvaluationError("GitHub Actions is disabled for the repository")
|
|
if allowed_actions not in {"all", "local_only", "selected"}:
|
|
raise EvaluationError("GitHub returned an invalid allowed_actions policy")
|
|
if not isinstance(sha_pinning, bool):
|
|
raise EvaluationError("GitHub returned an invalid SHA-pinning policy")
|
|
if allowed_actions != "selected":
|
|
return ActionsPolicy(True, str(allowed_actions), sha_pinning)
|
|
|
|
selected_url = f"{permissions_url}/selected-actions"
|
|
selected = _load_mapping(self.client, selected_url)
|
|
github_owned = selected.get("github_owned_allowed", False)
|
|
verified = selected.get("verified_allowed", False)
|
|
patterns = selected.get("patterns_allowed", [])
|
|
if not isinstance(github_owned, bool) or not isinstance(verified, bool):
|
|
raise EvaluationError("GitHub returned invalid selected-actions flags")
|
|
if not isinstance(patterns, list) or not all(
|
|
isinstance(pattern, str) for pattern in patterns
|
|
):
|
|
raise EvaluationError("GitHub returned invalid selected-actions patterns")
|
|
return ActionsPolicy(
|
|
True,
|
|
"selected",
|
|
sha_pinning,
|
|
github_owned,
|
|
verified,
|
|
tuple(pattern for pattern in patterns if isinstance(pattern, str)),
|
|
)
|
|
|
|
|
|
class EnterpriseClient:
|
|
ENTERPRISES_QUERY = """
|
|
query($cursor: String) {
|
|
viewer {
|
|
enterprises(first: 100, after: $cursor) {
|
|
nodes { id }
|
|
pageInfo { hasNextPage endCursor }
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
ORGANIZATIONS_QUERY = """
|
|
query($id: ID!, $cursor: String) {
|
|
node(id: $id) {
|
|
... on Enterprise {
|
|
organizations(first: 100, after: $cursor) {
|
|
nodes { login }
|
|
pageInfo { hasNextPage endCursor }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
def __init__(self, client: JsonClient) -> None:
|
|
self.client = client
|
|
|
|
def _query(self, query: str, variables: dict[str, object]) -> dict[str, object]:
|
|
value = self.client.post_json(
|
|
f"{API_ROOT}/graphql", {"query": query, "variables": variables}
|
|
)
|
|
response = _require_mapping(value, "GraphQL response")
|
|
if response.get("errors"):
|
|
raise EvaluationError("GitHub could not resolve enterprise ownership")
|
|
return _require_mapping(response.get("data"), "GraphQL response data")
|
|
|
|
def _enterprise_ids(self) -> list[str]:
|
|
enterprise_ids: list[str] = []
|
|
cursor: str | None = None
|
|
for _ in range(100):
|
|
data = self._query(self.ENTERPRISES_QUERY, {"cursor": cursor})
|
|
viewer = _require_mapping(data.get("viewer"), "GraphQL viewer")
|
|
connection = _require_mapping(
|
|
viewer.get("enterprises"), "GraphQL enterprises"
|
|
)
|
|
nodes = connection.get("nodes")
|
|
if not isinstance(nodes, list):
|
|
raise EvaluationError("GitHub returned invalid enterprise membership")
|
|
for node_value in nodes:
|
|
node = _require_mapping(node_value, "GraphQL enterprise")
|
|
enterprise_ids.append(_require_string(node, "id", "enterprise ID"))
|
|
page_info = _require_mapping(
|
|
connection.get("pageInfo"), "GraphQL enterprise page info"
|
|
)
|
|
if page_info.get("hasNextPage") is False:
|
|
return enterprise_ids
|
|
cursor = _require_string(page_info, "endCursor", "enterprise page cursor")
|
|
raise EvaluationError("enterprise membership exceeds pagination limit")
|
|
|
|
def _organizations(self, enterprise_id: str) -> frozenset[str]:
|
|
organizations: set[str] = set()
|
|
cursor: str | None = None
|
|
for _ in range(100):
|
|
data = self._query(
|
|
self.ORGANIZATIONS_QUERY, {"id": enterprise_id, "cursor": cursor}
|
|
)
|
|
enterprise = _require_mapping(data.get("node"), "GraphQL enterprise")
|
|
connection = _require_mapping(
|
|
enterprise.get("organizations"), "GraphQL enterprise organizations"
|
|
)
|
|
nodes = connection.get("nodes")
|
|
if not isinstance(nodes, list):
|
|
raise EvaluationError(
|
|
"GitHub returned invalid enterprise organizations"
|
|
)
|
|
for node_value in nodes:
|
|
node = _require_mapping(node_value, "GraphQL organization")
|
|
organizations.add(
|
|
_require_string(node, "login", "organization login").casefold()
|
|
)
|
|
page_info = _require_mapping(
|
|
connection.get("pageInfo"), "GraphQL organization page info"
|
|
)
|
|
if page_info.get("hasNextPage") is False:
|
|
return frozenset(organizations)
|
|
cursor = _require_string(page_info, "endCursor", "organization page cursor")
|
|
raise EvaluationError("enterprise organization list exceeds pagination limit")
|
|
|
|
def organizations_for(self, repository_owner: str) -> frozenset[str]:
|
|
owner = repository_owner.casefold()
|
|
for enterprise_id in self._enterprise_ids():
|
|
organizations = self._organizations(enterprise_id)
|
|
if owner in organizations:
|
|
return organizations
|
|
raise EvaluationError(
|
|
"enterprise-read token cannot resolve the repository owner's enterprise"
|
|
)
|
|
|
|
|
|
class MarketplaceClient:
|
|
def __init__(self, client: TextClient) -> None:
|
|
self.client = client
|
|
self.cache: dict[tuple[str, str], bool] = {}
|
|
|
|
def is_verified_action(self, owner: str, repository: str) -> bool:
|
|
key = (owner.casefold(), repository.casefold())
|
|
if key in self.cache:
|
|
return self.cache[key]
|
|
|
|
repository_url = f"{MARKETPLACE_ROOT}/{urllib.parse.quote(owner)}/{urllib.parse.quote(repository)}"
|
|
try:
|
|
repository_page = self.client.get_text(repository_url)
|
|
except HttpError as exc:
|
|
if exc.status == 404:
|
|
self.cache[key] = False
|
|
return False
|
|
raise
|
|
slugs = sorted(set(MARKETPLACE_LINK_RE.findall(html.unescape(repository_page))))
|
|
expected_prefix = f"{owner}/{repository}@"
|
|
prefix_marker = json.dumps(expected_prefix)
|
|
for slug in slugs:
|
|
listing_url = f"{MARKETPLACE_ROOT}/marketplace/actions/{slug}"
|
|
listing_page = self.client.get_text(listing_url)
|
|
marker = re.escape(f'"externalUsesPathPrefix":{prefix_marker}')
|
|
forward = re.findall(
|
|
rf'"isVerifiedOwner":(true|false)[^{{}}]{{0,2000}}{marker}',
|
|
listing_page,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
reverse = re.findall(
|
|
rf'{marker}[^{{}}]{{0,2000}}"isVerifiedOwner":(true|false)',
|
|
listing_page,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
verification_values = {value == "true" for value in [*forward, *reverse]}
|
|
if not verification_values and not re.search(
|
|
marker, listing_page, flags=re.IGNORECASE
|
|
):
|
|
continue
|
|
if len(verification_values) != 1:
|
|
raise EvaluationError(
|
|
f"Marketplace verification metadata is ambiguous for {owner}/{repository}"
|
|
)
|
|
verified = verification_values.pop()
|
|
self.cache[key] = verified
|
|
return verified
|
|
|
|
if re.search(re.escape(prefix_marker), repository_page, flags=re.IGNORECASE):
|
|
raise EvaluationError(
|
|
f"Marketplace verification metadata is ambiguous for {owner}/{repository}"
|
|
)
|
|
self.cache[key] = False
|
|
return False
|
|
|
|
|
|
class GitHubSourceClient:
|
|
def __init__(self, client: HttpClient) -> None:
|
|
self.client = client
|
|
|
|
def pull_files(
|
|
self, repository: str, number: int, expected_count: int
|
|
) -> list[dict[str, object]]:
|
|
files: list[dict[str, object]] = []
|
|
for page in range(1, 31):
|
|
url = (
|
|
f"{API_ROOT}/repos/{repository}/pulls/{number}/files"
|
|
f"?per_page=100&page={page}"
|
|
)
|
|
value = self.client.get_json(url)
|
|
if not isinstance(value, list) or not all(
|
|
isinstance(item, dict) for item in value
|
|
):
|
|
raise EvaluationError(
|
|
"GitHub returned an invalid pull-request file list"
|
|
)
|
|
batch = [_require_mapping(item, "pull-request file") for item in value]
|
|
files.extend(batch)
|
|
if len(batch) < 100:
|
|
if len(files) != expected_count:
|
|
raise EvaluationError(
|
|
"GitHub returned an incomplete pull-request file list "
|
|
f"({len(files)} of {expected_count} files)"
|
|
)
|
|
return files
|
|
raise EvaluationError(
|
|
"pull request exceeds GitHub's 3,000-file policy-evaluation limit"
|
|
)
|
|
|
|
def definition_paths(self, repository: str, revision: str) -> list[str]:
|
|
encoded_revision = urllib.parse.quote(revision, safe="")
|
|
url = f"{API_ROOT}/repos/{repository}/git/trees/{encoded_revision}?recursive=1"
|
|
value = self.client.get_json(url)
|
|
tree_data = _require_mapping(value, "Git tree")
|
|
if tree_data.get("truncated") is not False:
|
|
raise EvaluationError("GitHub returned a truncated merge-group tree")
|
|
tree = tree_data.get("tree")
|
|
if not isinstance(tree, list):
|
|
raise EvaluationError("GitHub returned an invalid merge-group tree")
|
|
paths: list[str] = []
|
|
total_size = 0
|
|
for entry_value in tree:
|
|
entry = _require_mapping(entry_value, "Git tree entry")
|
|
path = entry.get("path")
|
|
entry_type = entry.get("type")
|
|
if not isinstance(path, str) or not isinstance(entry_type, str):
|
|
raise EvaluationError("GitHub returned an invalid Git tree entry")
|
|
if entry_type == "blob" and is_action_definition(path):
|
|
size = entry.get("size")
|
|
if not isinstance(size, int) or size < 0:
|
|
raise EvaluationError(
|
|
"GitHub returned an invalid Git tree entry size"
|
|
)
|
|
paths.append(path)
|
|
total_size += size
|
|
if len(paths) > MAX_DEFINITION_COUNT:
|
|
raise EvaluationError("merge group has too many action definitions")
|
|
if total_size > MAX_TOTAL_DEFINITION_BYTES:
|
|
raise EvaluationError(
|
|
"merge-group action definitions exceed size limit"
|
|
)
|
|
return paths
|
|
|
|
def content(self, repository: str, path: str, revision: str) -> str:
|
|
encoded_path = urllib.parse.quote(path, safe="/")
|
|
encoded_revision = urllib.parse.quote(revision, safe="")
|
|
url = (
|
|
f"{API_ROOT}/repos/{repository}/contents/{encoded_path}"
|
|
f"?ref={encoded_revision}"
|
|
)
|
|
value = _require_mapping(self.client.get_json(url), "file content")
|
|
if value.get("encoding") != "base64":
|
|
raise EvaluationError(f"GitHub returned invalid file content for {path}")
|
|
encoded = value.get("content")
|
|
size = value.get("size", 0)
|
|
if not isinstance(encoded, str) or not isinstance(size, int):
|
|
raise EvaluationError(f"GitHub returned invalid file content for {path}")
|
|
if size > MAX_DEFINITION_BYTES:
|
|
raise EvaluationError(f"{path} exceeds the workflow policy size limit")
|
|
try:
|
|
decoded = base64.b64decode("".join(encoded.split()), validate=True)
|
|
return decoded.decode("utf-8")
|
|
except (ValueError, UnicodeDecodeError) as exc:
|
|
raise EvaluationError(
|
|
f"GitHub returned invalid file content for {path}"
|
|
) from exc
|
|
|
|
|
|
def collect_added_references(
|
|
source: GitHubSourceClient,
|
|
*,
|
|
repository: str,
|
|
pull_number: int,
|
|
base_repository: str,
|
|
base_sha: str,
|
|
head_repository: str,
|
|
head_sha: str,
|
|
changed_files: int,
|
|
) -> list[Reference]:
|
|
added: list[Reference] = []
|
|
for changed_file in source.pull_files(repository, pull_number, changed_files):
|
|
filename = changed_file.get("filename")
|
|
previous = changed_file.get("previous_filename", filename)
|
|
status = changed_file.get("status")
|
|
if not isinstance(filename, str) or not isinstance(previous, str):
|
|
raise EvaluationError("GitHub returned an invalid changed file entry")
|
|
if not is_action_definition(filename) and not is_action_definition(previous):
|
|
continue
|
|
|
|
base_text = ""
|
|
if status != "added" and is_action_definition(previous):
|
|
base_text = source.content(base_repository, previous, base_sha)
|
|
head_text = ""
|
|
if status != "removed" and is_action_definition(filename):
|
|
head_text = source.content(head_repository, filename, head_sha)
|
|
base_references = discover_references(base_text, previous)
|
|
head_references = discover_references(head_text, filename)
|
|
added.extend(added_references(base_references, head_references))
|
|
return added
|
|
|
|
|
|
def collect_all_references(
|
|
source: GitHubSourceClient, *, repository: str, revision: str
|
|
) -> list[Reference]:
|
|
references: list[Reference] = []
|
|
for path in source.definition_paths(repository, revision):
|
|
references.extend(
|
|
discover_references(source.content(repository, path, revision), path)
|
|
)
|
|
return references
|
|
|
|
|
|
def _require_mapping(value: object, label: str) -> dict[str, object]:
|
|
if not isinstance(value, dict):
|
|
raise EvaluationError(f"event is missing {label}")
|
|
return cast("dict[str, object]", value)
|
|
|
|
|
|
def _require_string(mapping: dict[str, object], key: str, label: str) -> str:
|
|
value = mapping.get(key)
|
|
if not isinstance(value, str) or not value:
|
|
raise EvaluationError(f"event is missing {label}")
|
|
return value
|
|
|
|
|
|
def _command_escape(value: str) -> str:
|
|
return (
|
|
value.replace("%", "%25")
|
|
.replace("\r", "%0D")
|
|
.replace("\n", "%0A")
|
|
.replace(":", "%3A")
|
|
.replace(",", "%2C")
|
|
)
|
|
|
|
|
|
def emit_error(message: str, reference: Reference | None = None) -> None:
|
|
if reference is None:
|
|
print(f"::error title=Workflow policy::{_command_escape(message)}")
|
|
return
|
|
print(
|
|
f"::error file={_command_escape(reference.path)},line={reference.line},"
|
|
f"title=Disallowed workflow reference::{_command_escape(message)}"
|
|
)
|
|
|
|
|
|
def append_summary(lines: list[str]) -> None:
|
|
summary_path = os.environ.get("GITHUB_STEP_SUMMARY")
|
|
if not summary_path:
|
|
return
|
|
with Path(summary_path).open("a", encoding="utf-8") as summary:
|
|
summary.write("\n".join(lines) + "\n")
|
|
|
|
|
|
class StatusPublisher:
|
|
def __init__(
|
|
self, client: HttpClient, repository: str, sha: str, target_url: str
|
|
) -> None:
|
|
self.client = client
|
|
self.url = f"{API_ROOT}/repos/{repository}/statuses/{sha}"
|
|
self.target_url = target_url
|
|
|
|
def publish(self, state: str, description: str) -> None:
|
|
payload = {
|
|
"state": state,
|
|
"context": STATUS_CONTEXT,
|
|
"description": description[:140],
|
|
"target_url": self.target_url,
|
|
}
|
|
self.client.post_json(self.url, payload)
|
|
print(f"{STATUS_CONTEXT}: {state} - {description}")
|
|
|
|
|
|
def evaluate_and_publish(
|
|
references: list[Reference],
|
|
*,
|
|
repository: str,
|
|
publisher: StatusPublisher,
|
|
empty_description: str,
|
|
) -> int:
|
|
policy_token = os.environ.get("ACTIONS_POLICY_READ_TOKEN", "")
|
|
enterprise_token = os.environ.get("ACTIONS_ENTERPRISE_READ_TOKEN", "")
|
|
if not references:
|
|
publisher.publish("success", empty_description)
|
|
append_summary(["## Workflow policy", f"{empty_description}."])
|
|
return 0
|
|
if not policy_token:
|
|
raise EvaluationError(
|
|
"ACTIONS_POLICY_READ_TOKEN is not configured; add a fine-grained "
|
|
"token with repository Administration (read)"
|
|
)
|
|
|
|
effective_policy = PolicyClient(HttpClient(policy_token)).get_policy(repository)
|
|
marketplace = MarketplaceClient(HttpClient())
|
|
repository_owner = repository.split("/", 1)[0]
|
|
enterprise_organizations: frozenset[str] = frozenset()
|
|
if effective_policy.allowed_actions == "selected":
|
|
if not enterprise_token:
|
|
raise EvaluationError(
|
|
"ACTIONS_ENTERPRISE_READ_TOKEN is not configured; add a classic "
|
|
"token with read:enterprise"
|
|
)
|
|
enterprise_organizations = EnterpriseClient(
|
|
HttpClient(enterprise_token)
|
|
).organizations_for(repository_owner)
|
|
rejected: list[tuple[Reference, str]] = []
|
|
for reference in references:
|
|
decision = evaluate_reference(
|
|
reference,
|
|
effective_policy,
|
|
repository_owner=repository_owner,
|
|
enterprise_organizations=enterprise_organizations,
|
|
is_verified_marketplace_action=marketplace.is_verified_action,
|
|
)
|
|
if not decision.allowed:
|
|
rejected.append((reference, decision.reason))
|
|
emit_error(f"{reference.value}: {decision.reason}", reference)
|
|
|
|
if rejected:
|
|
publisher.publish(
|
|
"failure", f"{len(rejected)} workflow reference(s) violate policy"
|
|
)
|
|
append_summary(
|
|
[
|
|
"## Workflow policy",
|
|
f"Rejected {len(rejected)} reference(s):",
|
|
*[
|
|
f"- `{reference.value}` in `{reference.path}:{reference.line}` — {reason}"
|
|
for reference, reason in rejected
|
|
],
|
|
]
|
|
)
|
|
return 1
|
|
|
|
publisher.publish("success", f"{len(references)} workflow reference(s) allowed")
|
|
append_summary(
|
|
[
|
|
"## Workflow policy",
|
|
f"All {len(references)} `uses:` references match the effective policy.",
|
|
]
|
|
)
|
|
return 0
|
|
|
|
|
|
def run_pull_request(
|
|
pull: dict[str, object],
|
|
*,
|
|
repository: str,
|
|
github_client: HttpClient,
|
|
target_url: str,
|
|
) -> int:
|
|
base = _require_mapping(pull.get("base"), "pull_request.base")
|
|
head = _require_mapping(pull.get("head"), "pull_request.head")
|
|
base_repo = _require_mapping(base.get("repo"), "pull_request.base.repo")
|
|
head_repo = _require_mapping(head.get("repo"), "pull_request.head.repo")
|
|
|
|
base_repository = _require_string(base_repo, "full_name", "base repository")
|
|
head_repository = _require_string(head_repo, "full_name", "head repository")
|
|
base_sha = _require_string(base, "sha", "base SHA")
|
|
head_sha = _require_string(head, "sha", "head SHA")
|
|
pull_number = pull.get("number")
|
|
changed_files = pull.get("changed_files")
|
|
if not isinstance(pull_number, int):
|
|
raise EvaluationError("event is missing pull request number")
|
|
if not isinstance(changed_files, int) or changed_files < 0:
|
|
raise EvaluationError("pull request has an invalid changed-files count")
|
|
|
|
publisher = StatusPublisher(github_client, repository, head_sha, target_url)
|
|
publisher.publish("pending", "Checking changed workflow references")
|
|
try:
|
|
references = collect_added_references(
|
|
GitHubSourceClient(github_client),
|
|
repository=repository,
|
|
pull_number=pull_number,
|
|
base_repository=base_repository,
|
|
base_sha=base_sha,
|
|
head_repository=head_repository,
|
|
head_sha=head_sha,
|
|
changed_files=changed_files,
|
|
)
|
|
return evaluate_and_publish(
|
|
references,
|
|
repository=repository,
|
|
publisher=publisher,
|
|
empty_description="No new or changed workflow references",
|
|
)
|
|
except EvaluationError as exc:
|
|
emit_error(str(exc))
|
|
publisher.publish("failure", "Workflow policy evaluation failed")
|
|
append_summary(["## Workflow policy", f"Evaluation failed: {exc}"])
|
|
return 1
|
|
|
|
|
|
def run_merge_group(
|
|
*,
|
|
repository: str,
|
|
head_sha: str,
|
|
github_client: HttpClient,
|
|
target_url: str,
|
|
) -> int:
|
|
publisher = StatusPublisher(github_client, repository, head_sha, target_url)
|
|
publisher.publish("pending", "Checking queued workflow references")
|
|
try:
|
|
references = collect_all_references(
|
|
GitHubSourceClient(github_client),
|
|
repository=repository,
|
|
revision=head_sha,
|
|
)
|
|
return evaluate_and_publish(
|
|
references,
|
|
repository=repository,
|
|
publisher=publisher,
|
|
empty_description="No workflow references in queued revision",
|
|
)
|
|
except EvaluationError as exc:
|
|
emit_error(str(exc))
|
|
publisher.publish("failure", "Merge-group policy evaluation failed")
|
|
append_summary(["## Workflow policy", f"Evaluation failed: {exc}"])
|
|
return 1
|
|
|
|
|
|
def _associated_pull_number(
|
|
workflow_run: dict[str, object],
|
|
*,
|
|
repository: str,
|
|
head_sha: str,
|
|
github_client: HttpClient,
|
|
) -> int:
|
|
pull_values = workflow_run.get("pull_requests")
|
|
numbers: set[int] = set()
|
|
if isinstance(pull_values, list):
|
|
for pull_value in pull_values:
|
|
pull = _require_mapping(pull_value, "workflow-run pull request")
|
|
number = pull.get("number")
|
|
if isinstance(number, int):
|
|
numbers.add(number)
|
|
if len(numbers) != 1:
|
|
pulls_url = (
|
|
f"{API_ROOT}/repos/{repository}/commits/{head_sha}/pulls?per_page=100"
|
|
)
|
|
associated = github_client.get_json(pulls_url)
|
|
if not isinstance(associated, list):
|
|
raise EvaluationError("GitHub returned invalid associated pull requests")
|
|
numbers = set()
|
|
for value in associated:
|
|
pull = _require_mapping(value, "associated pull request")
|
|
number = pull.get("number")
|
|
if isinstance(number, int):
|
|
numbers.add(number)
|
|
if len(numbers) != 1:
|
|
raise EvaluationError("workflow run is not bound to exactly one pull request")
|
|
return numbers.pop()
|
|
|
|
|
|
def run_workflow_run(
|
|
event_mapping: dict[str, object], github_client: HttpClient, target_url: str
|
|
) -> int:
|
|
repository_data = _require_mapping(event_mapping.get("repository"), "repository")
|
|
workflow_run = _require_mapping(event_mapping.get("workflow_run"), "workflow_run")
|
|
repository = _require_string(repository_data, "full_name", "repository.full_name")
|
|
head_sha = _require_string(workflow_run, "head_sha", "workflow-run head SHA")
|
|
trigger = _require_string(workflow_run, "event", "workflow-run event")
|
|
conclusion = workflow_run.get("conclusion")
|
|
workflow_path = _require_string(workflow_run, "path", "workflow-run path")
|
|
if workflow_path != ".github/workflows/workflow-policy-request.yml":
|
|
raise EvaluationError(f"unexpected workflow-run path {workflow_path!r}")
|
|
if trigger not in {"pull_request", "merge_group"}:
|
|
raise EvaluationError(f"unsupported workflow-run event {trigger!r}")
|
|
|
|
if trigger == "merge_group":
|
|
publisher = StatusPublisher(github_client, repository, head_sha, target_url)
|
|
if conclusion != "success":
|
|
publisher.publish("failure", "Workflow policy request did not succeed")
|
|
return 1
|
|
return run_merge_group(
|
|
repository=repository,
|
|
head_sha=head_sha,
|
|
github_client=github_client,
|
|
target_url=target_url,
|
|
)
|
|
|
|
pull_number = _associated_pull_number(
|
|
workflow_run,
|
|
repository=repository,
|
|
head_sha=head_sha,
|
|
github_client=github_client,
|
|
)
|
|
pull_url = f"{API_ROOT}/repos/{repository}/pulls/{pull_number}"
|
|
pull = _require_mapping(github_client.get_json(pull_url), "pull request")
|
|
head = _require_mapping(pull.get("head"), "pull_request.head")
|
|
current_head_sha = _require_string(head, "sha", "pull-request head SHA")
|
|
base = _require_mapping(pull.get("base"), "pull_request.base")
|
|
base_repo = _require_mapping(base.get("repo"), "pull_request.base.repo")
|
|
base_repository = _require_string(base_repo, "full_name", "base repository")
|
|
if base_repository.casefold() != repository.casefold():
|
|
raise EvaluationError("workflow run targets a different base repository")
|
|
if current_head_sha != head_sha:
|
|
print("Skipping stale workflow-policy request for an older pull-request head")
|
|
return 0
|
|
|
|
if conclusion != "success":
|
|
StatusPublisher(github_client, repository, head_sha, target_url).publish(
|
|
"failure", "Workflow policy request did not succeed"
|
|
)
|
|
return 1
|
|
return run_pull_request(
|
|
pull,
|
|
repository=repository,
|
|
github_client=github_client,
|
|
target_url=target_url,
|
|
)
|
|
|
|
|
|
def run() -> int:
|
|
event_path = os.environ.get("GITHUB_EVENT_PATH")
|
|
github_token = os.environ.get("GITHUB_TOKEN", "")
|
|
target_url = os.environ.get(
|
|
"GITHUB_RUN_URL", "https://github.com/NVIDIA/OpenShell/actions"
|
|
)
|
|
if not event_path or not github_token:
|
|
raise EvaluationError("GITHUB_EVENT_PATH and GITHUB_TOKEN are required")
|
|
try:
|
|
event = json.loads(Path(event_path).read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise EvaluationError(f"cannot read GitHub event: {exc}") from exc
|
|
event_mapping = _require_mapping(event, "payload")
|
|
github_client = HttpClient(github_token)
|
|
return run_workflow_run(event_mapping, github_client, target_url)
|
|
|
|
|
|
def main() -> int:
|
|
try:
|
|
return run()
|
|
except EvaluationError as exc:
|
|
emit_error(str(exc))
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|