forked from MakazhanAlpamys/Soup
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathci_workflow.py
More file actions
217 lines (187 loc) · 8.09 KB
/
Copy pathci_workflow.py
File metadata and controls
217 lines (187 loc) · 8.09 KB
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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
"""Render a GitHub Actions "fine-tuning gate" workflow (v0.71.35).
``soup ci init`` writes ``.github/workflows/soup-gate.yml`` — a CI job that
gates every PR on three Soup checks:
soup data validate <data> # dataset format compliance
soup expect <data> <suite> # expectations suite (PII / length / refusal)
soup ship --evidence <ev.json> # SHIP / DON'T-SHIP verdict (exit 2 = block)
The workflow body is a fixed YAML skeleton; only a handful of tokens are
interpolated. Following ``eval_gate_hook.py``: any value placed into a ``run:``
shell step is passed through ``shlex.quote`` (+ control-char rejection) so a
crafted path cannot chain commands, and every path is validated to stay under
the repo root. There is NO top-level torch import.
"""
from __future__ import annotations
import os
import re
import shlex
import tempfile
from typing import Optional
from soup_cli.utils.paths import enforce_under_cwd_and_no_symlink, is_under_cwd
# GitHub allows quite permissive ref names; we constrain to a safe subset that
# also cannot break the YAML scalar or a shell step.
_BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,254}$")
_PY_RE = re.compile(r"^\d+\.\d+$")
_MAX_PATH_LEN = 4096
_MAX_FILE_BYTES = 64 * 1024
# Chars that must never appear in a path rendered into a workflow. Beyond the
# obvious newlines, YAML 1.1 treats NEL/LS/PS as line breaks too, and a bare
# '#' starts a comment inside an unquoted plain scalar (truncating a run:
# step). Built via chr() so no raw separator ever sits in this source file.
_FORBIDDEN_PATH_CHARS = (chr(10), chr(13), chr(0x85), chr(0x2028), chr(0x2029), "#")
_WORKFLOW_TEMPLATE = """\
# Soup fine-tuning gate — generated by `soup ci init` (v0.71.35).
# Gates every PR on: data validate -> expectations -> SHIP verdict.
# Edit the paths below to match your repo. The `soup ship` step needs a
# committed evidence JSON (see `soup ship --evidence`); remove it if unused.
name: Soup Fine-tuning Gate
on:
push:
branches: [{branch}]
pull_request:
branches: [{branch}]
jobs:
soup-gate:
runs-on: ubuntu-latest
env:
PYTHONUTF8: "1"
PYTHONIOENCODING: "utf-8"
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "{python_version}"
- name: Install Soup CLI
# Core install is enough: `data validate`, `expect` and
# `ship --evidence` are all no-torch paths.
run: pip install soup-cli
- name: Validate training data
run: soup data validate {data}
- name: Run expectations suite
run: soup expect {data} {suite}
- name: SHIP / DON'T-SHIP gate
run: soup ship --evidence {evidence}{ship_config_arg}
"""
# Extra comment injected into the header when the gate is bound to a config —
# so the provenance/staleness behaviour is discoverable from the workflow.
_CONFIG_NOTE = (
"# The gate is bound to {config}: `soup ship` refuses evidence whose\n"
"# config_sha != this config's (produce it in your train job with\n"
"# `soup ship ... --config {config} --emit-evidence {evidence}`).\n"
)
def _validate_path(value: str, field: str) -> str:
if isinstance(value, bool) or not isinstance(value, str):
raise TypeError(f"{field} must be str")
if not value:
raise ValueError(f"{field} must be non-empty")
if "\x00" in value:
raise ValueError(f"{field} must not contain NUL")
# A newline / NEL / LS / PS or a '#' would break the value out of its
# single-line `run:` plain scalar (chaining a command or truncating the
# step); reject them all up front. See _FORBIDDEN_PATH_CHARS.
if any(ch in value for ch in _FORBIDDEN_PATH_CHARS):
raise ValueError(
f"{field} must be a single line and must not contain '#'"
)
if len(value) > _MAX_PATH_LEN:
raise ValueError(f"{field} exceeds {_MAX_PATH_LEN} characters")
if not is_under_cwd(value):
raise ValueError(f"{field} must stay under the repository root")
return value
def _safe_shell_quote(value: str) -> str:
"""``shlex.quote`` with a control-char rejection prelude (defence in depth)."""
if any(ord(ch) < 0x20 for ch in value):
raise ValueError("value contains control characters")
return shlex.quote(value)
def render_soup_gate_workflow(
*,
data_path: str,
suite_path: str,
evidence_path: str,
python_version: str = "3.11",
branch: str = "main",
config_path: Optional[str] = None,
) -> str:
"""Render the workflow YAML body. Deterministic, no I/O.
Every path is validated to stay under the repo root and shell-quoted before
it reaches a ``run:`` step; ``python_version`` / ``branch`` are regex-gated.
When ``config_path`` is given, the ``soup ship`` step is bound to that
committed config (``--config``), which makes the gate refuse evidence whose
``config_sha`` drifted from the config in the PR (v0.71.39 provenance/
staleness) — the committed evidence must describe the committed recipe.
"""
data = _validate_path(data_path, "data_path")
suite = _validate_path(suite_path, "suite_path")
evidence = _validate_path(evidence_path, "evidence_path")
if not isinstance(python_version, str) or not _PY_RE.match(python_version):
raise ValueError("python_version must look like '3.11'")
if not isinstance(branch, str) or not _BRANCH_RE.match(branch):
raise ValueError("branch has an unsupported name")
ship_config_arg = ""
header = ""
if config_path is not None:
cfg = _validate_path(config_path, "config_path")
ship_config_arg = f" --config {_safe_shell_quote(cfg)}"
header = _CONFIG_NOTE.format(config=cfg, evidence=evidence)
body = _WORKFLOW_TEMPLATE.format(
branch=branch,
python_version=python_version,
data=_safe_shell_quote(data),
suite=_safe_shell_quote(suite),
evidence=_safe_shell_quote(evidence),
ship_config_arg=ship_config_arg,
)
return header + body
def write_soup_gate_workflow(
*,
data_path: str,
suite_path: str,
evidence_path: str,
python_version: str = "3.11",
branch: str = "main",
output_path: str = ".github/workflows/soup-gate.yml",
overwrite: bool = False,
config_path: Optional[str] = None,
) -> str:
"""Render + atomically write the workflow. Returns the path written.
Refuses to clobber an existing file unless ``overwrite=True``; rejects a
symlink destination (TOCTOU) and any path outside the repo root.
"""
body = render_soup_gate_workflow(
data_path=data_path,
suite_path=suite_path,
evidence_path=evidence_path,
python_version=python_version,
branch=branch,
config_path=config_path,
)
if isinstance(output_path, bool) or not isinstance(output_path, str):
raise TypeError("output_path must be str")
if not output_path:
raise ValueError("output_path must be non-empty")
if "\x00" in output_path:
raise ValueError("output_path must not contain NUL")
if not isinstance(overwrite, bool):
raise TypeError("overwrite must be bool")
# Single-source-of-truth containment + symlink/reparse-point rejection.
# A hand-rolled S_ISLNK check misses Windows junctions, so always delegate
# to the shared helper (v0.71.35 security review).
enforce_under_cwd_and_no_symlink(output_path, "output_path")
if os.path.lexists(output_path) and not overwrite:
raise ValueError("workflow already exists; pass --force to overwrite")
if len(body.encode("utf-8")) > _MAX_FILE_BYTES:
raise ValueError("rendered workflow exceeds 64 KiB cap")
parent = os.path.dirname(os.path.abspath(output_path)) or "."
os.makedirs(parent, exist_ok=True)
fd, tmp = tempfile.mkstemp(prefix=".soup-gate.", dir=parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(body)
os.replace(tmp, output_path)
except Exception:
try:
os.unlink(tmp)
except OSError:
pass
raise
return output_path
__all__ = ["render_soup_gate_workflow", "write_soup_gate_workflow"]