Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264"""Shell lines read as run() calls, and the guard that keeps cells on run().
Cells start programs with run(program, *args): no shell, supervised, traced.A shell line or a raw process API from a cell is refused with the run() callit means when the line is simple enough to read, so the refusal teaches byexample. This is guidance, not a sandbox: libraries and plugins still spawn."""
from __future__ import annotations
from collections.abc import Sequenceimport osimport reimport shleximport sysimport sysconfig
SHELLS = {"sh", "bash", "zsh", "dash", "ksh", "fish", "nu"}CELL_PREFIX = "<albedo:" # the filename every cell compiles underSPAWN_EVENTS = { "subprocess.Popen", "os.system", "os.posix_spawn", "os.exec", "os.spawn",}# stdlib frames sit between a cell and the spawn it asked for; site-packages do notSTDLIB = tuple({sysconfig.get_paths()[key] for key in ("stdlib", "platstdlib")})HINTS = ( "cd dir && … → cwd=, NAME=value → env=, a | b → run(a…).pipe(b…), " "| tail -n N → .tail(lines=N), | head -n N → .head(lines=N), 2>&1 is implied, " "and chains, loops, and globs are Python over the job's output.")
class Refused(PermissionError): """A shell or raw process API where run() belongs."""
def shell_script(argv: Sequence[str]) -> str | None: """The script of `sh -c script` and its spellings (`bash -lc`), else None.""" if not argv or os.path.basename(argv[0]) not in SHELLS: return None for index, arg in enumerate(argv[1:], 1): if arg.startswith("--"): continue # --norc, --login if not arg.startswith("-"): return None if "c" in arg[1:]: return argv[index + 1] if index + 1 < len(argv) else "" return None
def words(items: Sequence[object]) -> list[str]: """run()'s arguments as argv: text, paths, and numbers.""" argv = [] for item in items: if isinstance(item, (str, bytes, os.PathLike)): argv.append(os.fsdecode(item)) elif isinstance(item, (int, float)) and not isinstance(item, bool): argv.append(str(item)) else: raise TypeError( f"run() arguments are text, paths, or numbers, not {type(item).__name__}" ) return argv
def call( argv: Sequence[str], cwd: str | None = None, env: dict[str, str] | None = None) -> str: """argv as the run() call that starts it.""" parts = [repr(arg) for arg in argv] if cwd: parts.append(f"cwd={cwd!r}") if env: parts.append(f"env={env!r}") return f"run({', '.join(parts)})"
def _unquoted(script: str) -> str: """script with quoted text blanked, so shell syntax outside quotes shows.""" out, quote, escaped = [], "", False for char in script: if escaped: out.append(" ") escaped = False elif char == "\\" and quote != "'": out.append(" ") escaped = True elif quote: out.append(" ") quote = "" if char == quote else quote elif char in "'\"": out.append(" ") quote = char else: out.append(char) return "".join(out)
_LIMIT = re.compile(r"^(head|tail)(?:\s+-n\s*(\d+)|\s+-(\d+))?$")
def _split(text: str, plain: str, separator: str) -> list[tuple[str, str]]: """text, and its unquoted form, cut where separator stands outside quotes.""" parts, start = [], 0 for match in re.finditer(separator, plain): parts.append((text[start : match.start()], plain[start : match.start()])) start = match.end() parts.append((text[start:], plain[start:])) return [part for part in parts if part[0].strip()]
def _stage(text: str) -> tuple[list[str], dict[str, str]]: """One pipeline stage's argv and the NAME=value words before it.""" words, env = shlex.split(text), {} while words and re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", words[0]): name, _, value = words.pop(0).partition("=") env[name] = value if not words: raise ValueError("a stage with no program") return words, env
def translate(script: str) -> str | None: """The run() code a simple shell line means, or None when it does more than start programs: `cd cli && go test ./... | tail -5` is `job = run('go', 'test', './...', cwd='cli')`, `await job`, then `job.tail(lines=5)`, and each other `|` is a .pipe(...).""" plain = _unquoted(script) for match in re.finditer(r"2>&1", plain): # stderr joins stdout anyway script = script[: match.start()] + " " + script[match.end() :] plain = plain.replace("2>&1", " ") if ( re.search(r"[$`*?<>(){}&]|(?:^|\s)~", plain.replace("&&", " ")) or "||" in plain ): return None cwd, stages = None, None try: for text, part in _split(script, plain, r"&&|;|\n"): piped = [stage for stage, _ in _split(text, part, r"\|")] words = shlex.split(piped[0]) if words[:1] == ["cd"] and len(words) == 2 and stages is None: cwd = words[1] elif stages is None: stages = piped else: return None if stages is None: return None limit = _LIMIT.match(stages[-1].strip()) if len(stages) > 1 else None if limit is not None: stages = stages[:-1] calls = [] for index, stage in enumerate(stages): words, env = _stage(stage) inner = shell_script(words) if inner is not None: # a shell inside the line: read what it runs return ( translate(inner) if len(stages) == 1 and limit is None and not env else None ) call_text = call(words, cwd, env) calls.append( call_text if index == 0 else ".pipe" + call_text.removeprefix("run") ) except ValueError: return None read = ( "tail()" if limit is None else f"{limit.group(1)}(lines={limit.group(2) or limit.group(3) or 10})" ) return f"job = {''.join(calls)}\nawait job\njob.{read}"
def refusal( what: str, script: str | None = None, argv: Sequence[str] | None = None, cwd: str | None = None,) -> Refused: """Why `what` is refused, with the run() call it means when there is one.""" suggestion = ( translate(script) if script is not None else (f"job = await {call(argv, cwd)}\njob.tail()" if argv else None) ) if suggestion: return Refused( f"{what} is refused here; run() starts programs without a shell. this one is:\n" + "".join(f" {line}\n" for line in suggestion.splitlines()) ) return Refused( f"{what} is refused here; run() starts one program without a shell: " f"run('git', 'status', cwd='cli'). {HINTS}" )
installed = False
def _from_cell(skip: int) -> bool: """Whether the spawn under way was asked for by cell code itself, rather than by a plugin or library the cell called. skip counts the audit hook's own frames above the spawning call.""" try: frame = sys._getframe(skip) except ValueError: return False while frame is not None: name = frame.f_code.co_filename if name.startswith(CELL_PREFIX): return True stdlib = name.startswith("<frozen") or ( name.startswith(STDLIB) and "-packages" not in name ) if not stdlib: return False frame = frame.f_back return False
def _strings(args: object) -> list[str]: if isinstance(args, (str, bytes, os.PathLike)): return [os.fsdecode(args)] if isinstance(args, (list, tuple)): return [ os.fsdecode(arg) if isinstance(arg, (str, bytes, os.PathLike)) else str(arg) for arg in args ] return []
def guard(event: str, args: tuple[object, ...]) -> None: """Audit hook: a cell that spawns through subprocess or os gets run() instead.""" if event not in SPAWN_EVENTS or not _from_cell(skip=2): return if event == "os.system": raise refusal("os.system", script=_strings(args[0])[0]) argv = _strings(args[2] if event == "os.spawn" else args[1]) cwd = ( _strings(args[2])[0] if event == "subprocess.Popen" and args[2] is not None else None ) script = shell_script(argv) what = "subprocess" if event == "subprocess.Popen" else event raise ( refusal(what, script=script) if script is not None else refusal(what, argv=argv, cwd=cwd) )
def install() -> None: global installed if not installed: installed = True sys.addaudithook(guard)