Spaces:
Running
Running
File size: 5,147 Bytes
da5cba1 599974a da5cba1 | 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 | """Running a Dockerfile-only task on an HF Sandbox, which starts from a prebuilt image and builds nothing.
Most task Dockerfiles are one stage: FROM an image, then RUN, COPY, WORKDIR, ENV and ARG. Those can be replayed in
the sandbox itself: start it from the FROM image, run each RUN there, upload each COPY from the task's environment
folder, and keep ENV and WORKDIR for every later command (the agent's and the grader's too). What can't be replayed
(several stages, COPY --from, heredocs, ADD from a URL) is refused with the reason, so the run panel can say why.
"""
from __future__ import annotations
import json
import re
import shlex
from dataclasses import dataclass, field
IGNORED = {"CMD", "ENTRYPOINT", "EXPOSE", "LABEL", "HEALTHCHECK", "VOLUME", "STOPSIGNAL", "MAINTAINER"}
@dataclass
class Plan:
base: str
steps: list[tuple[str, object]] = field(default_factory=list) # ("run", cmd) ("copy", (srcs, dst)) ("workdir", p) ("env", {k: v})
class NotReplayable(ValueError):
pass
def _lines(text: str) -> list[str]:
"""Instructions, one per entry: comments dropped, continuation lines joined."""
out, cur = [], ""
for raw in text.splitlines():
line = raw.rstrip()
if not cur and (not line.strip() or line.lstrip().startswith("#")):
continue
if cur and line.lstrip().startswith("#"): # a comment inside a continued instruction
continue
if line.endswith("\\"):
cur += line[:-1] + " "
continue
cur += line
out.append(cur.strip())
cur = ""
if cur.strip():
out.append(cur.strip())
return out
def _subst(s: str, args: dict[str, str]) -> str:
return re.sub(r"\$\{(\w+)(?::-([^}]*))?\}|\$(\w+)",
lambda m: args.get(m.group(1) or m.group(3), m.group(2) or "") if (m.group(1) or m.group(3)) in args or m.group(2) is not None
else m.group(0), s)
def _pairs(rest: str) -> dict[str, str]:
"""ENV/ARG bodies: `K=V K2="v 2"` or the old `K V` form."""
parts = shlex.split(rest)
if parts and "=" not in parts[0]:
return {parts[0]: " ".join(parts[1:])}
return dict(p.split("=", 1) if "=" in p else (p, "") for p in parts)
def plan(text: str) -> Plan:
"""The replay plan for a Dockerfile, or NotReplayable with the reason."""
if re.search(r"<<-?\s*['\"]?\w+", text):
raise NotReplayable("its Dockerfile uses a heredoc")
args: dict[str, str] = {}
base = None
steps: list[tuple[str, object]] = []
for line in _lines(text):
op, _, rest = line.partition(" ")
op, rest = op.upper(), rest.strip()
if op == "FROM":
if base:
raise NotReplayable("its Dockerfile has several stages")
base = _subst(rest.split()[0] if not rest.startswith("--") else rest.split()[1], args)
continue
if op == "ARG":
for k, v in _pairs(rest).items():
args.setdefault(k, v)
continue
if base is None:
raise NotReplayable("its Dockerfile has no FROM")
rest = _subst(rest, args)
if op == "RUN":
if rest.startswith("--"):
flags = re.match(r"((?:--\S+\s+)+)(.*)", rest, re.S)
rest = flags.group(2) if flags else rest
if rest.startswith("["):
try:
rest = " ".join(shlex.quote(x) for x in json.loads(rest))
except ValueError:
raise NotReplayable("a RUN line it can't read")
steps.append(("run", rest))
elif op in ("COPY", "ADD"):
parts = [p for p in (json.loads(rest) if rest.startswith("[") else shlex.split(rest))]
flags = [p for p in parts if p.startswith("--")]
if any(f.startswith("--from") for f in flags):
raise NotReplayable("its Dockerfile copies from another stage")
parts = [p for p in parts if not p.startswith("--")]
if len(parts) < 2:
raise NotReplayable("a COPY line it can't read")
if any(re.match(r"https?://", p) for p in parts[:-1]):
raise NotReplayable("its Dockerfile adds a file from a URL")
steps.append(("copy", (parts[:-1], parts[-1])))
elif op == "WORKDIR":
steps.append(("workdir", rest))
elif op == "ENV":
steps.append(("env", _pairs(rest)))
elif op in IGNORED or op == "ONBUILD":
continue
else:
raise NotReplayable(f"its Dockerfile uses {op}")
if not base:
raise NotReplayable("its Dockerfile has no FROM")
if base.lower() == "scratch":
raise NotReplayable("its Dockerfile starts FROM scratch")
return Plan(base=base, steps=steps)
def check(text: str | None) -> tuple[bool, str]:
"""Whether a task with this Dockerfile can run here, and if not, why."""
if not text:
return False, "it has neither a prebuilt image nor a Dockerfile"
try:
plan(text)
return True, ""
except NotReplayable as e:
return False, str(e)
|