Spaces:
Running
Running
File size: 6,377 Bytes
3755ae7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | from __future__ import annotations
import unittest
from pathlib import Path
import yaml
from yaml.nodes import MappingNode, Node, ScalarNode, SequenceNode
ROOT = Path(__file__).resolve().parents[1]
WORKFLOW_DIRECTORY = ROOT / ".github" / "workflows"
FULL_COMMIT_SHA_LENGTH = 40
def mapping_values(node: Node, name: str) -> list[Node]:
if not isinstance(node, MappingNode):
return []
return [
value
for key, value in node.value
if isinstance(key, ScalarNode) and key.value == name
]
def uses_references(node: Node) -> list[tuple[int, str]]:
references: list[tuple[int, str]] = []
if not isinstance(node, MappingNode):
return references
for key, value in node.value:
if isinstance(key, ScalarNode) and key.value == "uses":
reference = value.value if isinstance(value, ScalarNode) else "<non-scalar>"
references.append((key.start_mark.line + 1, reference))
return references
def step_action_references(node: Node, active_nodes: set[int] | None = None) -> list[tuple[int, str]]:
if not isinstance(node, MappingNode):
return []
active = set() if active_nodes is None else active_nodes
identity = id(node)
if identity in active:
return []
active.add(identity)
references = uses_references(node)
for parallel in mapping_values(node, "parallel"):
if isinstance(parallel, SequenceNode):
for child in parallel.value:
references.extend(step_action_references(child, active))
active.remove(identity)
return references
def action_references(document: Node) -> list[tuple[int, str]]:
references: list[tuple[int, str]] = []
for jobs in mapping_values(document, "jobs"):
if not isinstance(jobs, MappingNode):
continue
for _, job in jobs.value:
references.extend(uses_references(job))
for steps in mapping_values(job, "steps"):
if isinstance(steps, SequenceNode):
for step in steps.value:
references.extend(step_action_references(step))
return references
def mutable_external_action_references(workflow: str) -> list[tuple[int, str]]:
mutable: list[tuple[int, str]] = []
for document in yaml.compose_all(workflow):
if document is None:
continue
for line_number, reference in action_references(document):
if reference.startswith(("./", "$/", "docker://")):
continue
action, separator, revision = reference.rpartition("@")
components = action.split("/")
full_sha = len(revision) == FULL_COMMIT_SHA_LENGTH and all(
character in "0123456789abcdefABCDEF" for character in revision
)
if (
not separator
or len(components) < 2
or any(not part for part in components)
or not full_sha
):
mutable.append((line_number, reference))
return mutable
class WorkflowSecurityTests(unittest.TestCase):
def test_repository_workflows_pin_external_actions_to_full_commit_shas(self) -> None:
workflows = sorted(WORKFLOW_DIRECTORY.glob("*.yml"))
workflows.extend(sorted(WORKFLOW_DIRECTORY.glob("*.yaml")))
self.assertTrue(workflows)
violations = {
path.relative_to(ROOT).as_posix(): mutable_external_action_references(
path.read_text(encoding="utf-8")
)
for path in workflows
}
self.assertEqual({}, {path: refs for path, refs in violations.items() if refs})
def test_mutable_external_action_tags_are_rejected_across_yaml_styles(self) -> None:
workflow = (
"jobs:\n"
" inline: {uses: noqt/reusable/.github/workflows/check.yml@main}\n"
" quoted:\n"
' steps: [{"uses": actions/checkout@v6}]\n'
)
self.assertEqual(
[
(2, "noqt/reusable/.github/workflows/check.yml@main"),
(4, "actions/checkout@v6"),
],
mutable_external_action_references(workflow),
)
def test_mutable_action_in_parallel_step_group_is_rejected(self) -> None:
workflow = (
"jobs:\n"
" test:\n"
" steps:\n"
" - parallel:\n"
" - uses: actions/checkout@v6\n"
" - uses: example/action@0123456789abcdef0123456789abcdef01234567\n"
)
self.assertEqual(
[(5, "actions/checkout@v6")],
mutable_external_action_references(workflow),
)
def test_full_commit_sha_local_action_and_script_text_are_allowed(self) -> None:
workflow = (
"jobs:\n"
" reusable:\n"
" uses: noqt/reusable/.github/workflows/check.yml@0123456789abcdef0123456789abcdef01234567\n"
" same-repository:\n"
" uses: $/.github/workflows/check.yml\n"
" test:\n"
" steps:\n"
" - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6\n"
" - uses: ./.github/actions/local-check\n"
" - uses: $/.github/actions/local-check\n"
" - run: |\n"
" printf 'uses: actions/checkout@v6'\n"
" - env: {uses: actions/checkout@v6}\n"
" run: echo safe\n"
)
self.assertEqual([], mutable_external_action_references(workflow))
def test_only_exactly_40_hexadecimal_revision_characters_are_allowed(self) -> None:
revisions = {
"39": "a" * 39,
"40": "b" * 40,
"41": "c" * 41,
"non-hex": "g" * 40,
}
workflow = "jobs:\n test:\n steps:\n" + "".join(
f" - uses: example/action@{revision}\n" for revision in revisions.values()
)
self.assertEqual(
[
(4, f"example/action@{revisions['39']}"),
(6, f"example/action@{revisions['41']}"),
(7, f"example/action@{revisions['non-hex']}"),
],
mutable_external_action_references(workflow),
)
if __name__ == "__main__":
unittest.main()
|