64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
|
"""Opening and downloading run artefacts, safely.
|
||
|
|
|
||
|
|
An artefact is a *reference* to something a run produced: a file path or a URL held in a datapoint. The UI may serve a local
|
||
|
|
file only if it is exactly the artefact's resolved location, is a regular file of reasonable size, is not in a sensitive place,
|
||
|
|
and does not look like it contains a secret. Anything else is refused with a reason.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import mimetypes
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from .pack import SECRET_RE
|
||
|
|
|
||
|
|
MAX_BYTES = 5 * 1024 * 1024
|
||
|
|
SENSITIVE_PARTS = {".ssh", ".gnupg", ".aws", ".kube", ".git", ".password-store", "runbook-runs", "sops", ".docker"}
|
||
|
|
SENSITIVE_SUFFIXES = (".pem", ".key", ".p12", ".pfx", ".kdbx", ".age")
|
||
|
|
SENSITIVE_NAMES = {"shadow", "gshadow", "id_rsa", "id_ed25519", "keys.txt", ".env", ".netrc", ".pgpass"}
|
||
|
|
|
||
|
|
|
||
|
|
def is_url(location: str) -> bool:
|
||
|
|
return location.startswith(("https://", "http://"))
|
||
|
|
|
||
|
|
|
||
|
|
def check(location: str | None) -> tuple[Path | None, str]:
|
||
|
|
"""(path, '') if the file may be served, else (None, reason)."""
|
||
|
|
if not location:
|
||
|
|
return None, "the location is not known yet"
|
||
|
|
if is_url(location):
|
||
|
|
return None, "this artefact is a link, not a local file"
|
||
|
|
p = Path(location).expanduser()
|
||
|
|
if not p.is_absolute():
|
||
|
|
return None, "only absolute file locations can be opened"
|
||
|
|
try:
|
||
|
|
p = p.resolve()
|
||
|
|
except OSError:
|
||
|
|
return None, "the location cannot be resolved"
|
||
|
|
if not p.exists():
|
||
|
|
return None, "the file does not exist (yet)"
|
||
|
|
if not p.is_file():
|
||
|
|
return None, "the location is not a regular file"
|
||
|
|
parts = {x.lower() for x in p.parts}
|
||
|
|
if parts & SENSITIVE_PARTS or p.name.lower() in SENSITIVE_NAMES or p.name.lower().endswith(SENSITIVE_SUFFIXES):
|
||
|
|
return None, "this file is in a sensitive place and is never served"
|
||
|
|
try:
|
||
|
|
size = p.stat().st_size
|
||
|
|
if size > MAX_BYTES:
|
||
|
|
return None, f"the file is larger than {MAX_BYTES // (1024 * 1024)} MB"
|
||
|
|
head = p.read_bytes()[: 256 * 1024]
|
||
|
|
except OSError as e:
|
||
|
|
return None, f"the file cannot be read: {e.strerror}"
|
||
|
|
if SECRET_RE.search(head.decode("utf-8", "ignore")):
|
||
|
|
return None, "the file looks like it contains a secret and is never served"
|
||
|
|
return p, ""
|
||
|
|
|
||
|
|
|
||
|
|
def content(path: Path) -> tuple[bytes, str, bool]:
|
||
|
|
"""(bytes, content type, is_text). Text is served as plain text; everything else only as a download."""
|
||
|
|
data = path.read_bytes()
|
||
|
|
try:
|
||
|
|
data.decode("utf-8")
|
||
|
|
return data, "text/plain; charset=utf-8", True
|
||
|
|
except UnicodeDecodeError:
|
||
|
|
return data, mimetypes.guess_type(path.name)[0] or "application/octet-stream", False
|