Repository navigation
Expand file tree
/
Copy pathshell_command.py
More file actions
2071 lines (1754 loc) · 78.8 KB
/
Copy pathshell_command.py
File metadata and controls
2071 lines (1754 loc) · 78.8 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
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
from __future__ import annotations
import asyncio
import os
import re
import shlex
import shutil
import sys
import time
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
from typing import NamedTuple
from loguru import logger
from pydantic_ai import ApprovalRequired, RunContext, Tool
from .. import constants as cs
from .. import logs as ls
from .. import tool_errors as te
from ..config import settings
from ..decorators import async_timing_decorator
from ..schemas import ShellCommandResult
from ..taint import ReadContentRecord
from . import tool_descriptions as td
PIPELINE_PATTERNS_COMPILED = tuple(
(re.compile(pattern, re.IGNORECASE), reason)
for pattern, reason in cs.SHELL_DANGEROUS_PATTERNS_PIPELINE
)
SEGMENT_PATTERNS_COMPILED = tuple(
(re.compile(pattern, re.IGNORECASE), reason)
for pattern, reason in cs.SHELL_DANGEROUS_PATTERNS_SEGMENT
)
def _is_outside_single_quotes(command: str, pos: int) -> bool:
in_single = False
i = 0
while i < pos:
char = command[i]
if char == "\\" and not in_single and i + 1 < len(command):
i += 2
continue
if char == "'":
in_single = not in_single
i += 1
return not in_single
def _has_subshell(command: str) -> str | None:
for pattern in cs.SHELL_SUBSHELL_PATTERNS:
start = 0
while True:
pos = command.find(pattern, start)
if pos == -1:
break
if _is_outside_single_quotes(command, pos):
return pattern
start = pos + 1
return None
class CommandGroup:
__slots__ = ("commands", "operator")
def __init__(self, commands: list[str], operator: str | None = None):
self.commands = commands
self.operator = operator
@dataclass
class _CommandParser:
"""Character-level state of `_parse_command` while it walks a command."""
command: str
groups: list[CommandGroup] = field(default_factory=list)
current_pipeline: list[str] = field(default_factory=list)
current_segment: list[str] = field(default_factory=list)
in_single: bool = False
in_double: bool = False
pending_operator: str | None = None
def finalize_segment(self) -> None:
seg = "".join(self.current_segment).strip()
if seg:
self.current_pipeline.append(seg)
self.current_segment.clear()
def finalize_group(self, new_operator: str) -> None:
self.finalize_segment()
if self.current_pipeline:
self.groups.append(
CommandGroup(list(self.current_pipeline), self.pending_operator)
)
self.current_pipeline.clear()
self.pending_operator = new_operator
def step(self, i: int) -> int:
"""Consume the character at `i` and return the index of the next one."""
command = self.command
char = command[i]
if char == "\\" and i + 1 < len(command):
self.current_segment.append(char)
self.current_segment.append(command[i + 1])
return i + 2
if char == "'" and not self.in_double:
self.in_single = not self.in_single
self.current_segment.append(char)
elif char == '"' and not self.in_single:
self.in_double = not self.in_double
self.current_segment.append(char)
elif char in ("|", "&", ";") and not self.in_single and not self.in_double:
return self._operator_step(i)
else:
self.current_segment.append(char)
return i + 1
def _operator_step(self, i: int) -> int:
"""Handle an unquoted `|`, `&` or `;` at `i`; return the next index."""
command = self.command
char = command[i]
if char == ";":
self.finalize_group(";")
elif char == "|":
if i + 1 < len(command) and command[i + 1] == "|":
self.finalize_group("||")
return i + 2
self.finalize_segment()
else:
if i + 1 < len(command) and command[i + 1] == "&":
self.finalize_group("&&")
return i + 2
self.current_segment.append(char)
return i + 1
def _parse_command(command: str) -> list[CommandGroup]:
parser = _CommandParser(command)
i = 0
while i < len(command):
i = parser.step(i)
parser.finalize_segment()
if parser.current_pipeline:
parser.groups.append(
CommandGroup(list(parser.current_pipeline), parser.pending_operator)
)
return parser.groups
def _is_blocked_command(cmd: str) -> bool:
return cmd in cs.SHELL_DANGEROUS_COMMANDS
def _is_dangerous_rm(cmd_parts: list[str]) -> bool:
if not cmd_parts or cmd_parts[0] != cs.SHELL_CMD_RM:
return False
flags = "".join(_rm_options(cmd_parts[1:]))
return "r" in flags and "f" in flags
def _rm_options(args: list[str]) -> list[str]:
"""The tokens any supported rm might parse as options.
Deliberately the mirror image of `_rm_operands`. GNU rm permutes, so it
honours a flag written after an operand: `rm target -rf` deletes
recursively and without prompting. Stopping at the first operand, as the
POSIX operand rule does, would miss that and let the flag check pass.
Only `--` ends option parsing here, because after it a dash-prefixed token
is a filename on every rm. That keeps `rm -r -- -in-root-file` off the
dangerous-flag path, which is the case that made reading whole operands as
flags wrong in the first place.
Each helper errs towards catching more: operands POSIX-maximal so more
paths face the containment check, options GNU-maximal so more spellings
face the flag check.
"""
end = args.index("--") if "--" in args else len(args)
return [part for part in args[:end] if part.startswith("-") and part != "-"]
def _rm_operands(args: list[str]) -> list[str]:
"""The tokens rm would treat as paths.
POSIX option parsing ends at the first operand or at `--`, whichever comes
first, and everything from there on is a path even when it starts with a
dash. BSD rm (macOS) follows that literally, so `rm a -x/../../outside/x`
deletes the second token; GNU rm instead permutes and rejects `-x/` as a
bad option. Taking the POSIX reading treats more tokens as paths than GNU
would, so the containment check only ever errs towards refusing.
Dropping every dash-prefixed token, as this used to, let both
`rm -r -- -x/../../outside/victim` and `rm a -x/../../outside/victim`
skip the containment check entirely.
"""
for index, token in enumerate(args):
if token == "--":
return args[index + 1 :]
if token == "-" or not token.startswith("-"):
return args[index:]
return []
def _is_dangerous_rm_path(cmd_parts: list[str], project_root: Path) -> tuple[bool, str]:
if not cmd_parts or cmd_parts[0] != cs.SHELL_CMD_RM:
return False, ""
for path_arg in _rm_operands(cmd_parts[1:]):
if (reason := _dangerous_rm_target(path_arg, project_root)) is not None:
return True, reason
return False, ""
def _dangerous_rm_target(path_arg: str, project_root: Path) -> str | None:
if path_arg in ("*", ".", ".."):
return f"rm targeting dangerous path: {path_arg}"
# Joining onto the root is the platform's own absoluteness test: an
# absolute target replaces the root outright, a relative one lands under
# it. A `startswith("/")` check is POSIX-only: on Windows a rooted,
# drive-less target such as `/x` would resolve on the Python process's
# drive rather than the root's, the drive rm runs on.
try:
resolved = (project_root / path_arg).resolve()
except (OSError, ValueError):
return f"rm with invalid path: {path_arg}"
if resolved == project_root:
return "rm targeting the project root"
resolved_str = str(resolved)
if resolved == resolved.parent:
return "rm targeting root directory"
try:
resolved.relative_to(project_root)
except ValueError:
parts = resolved.parts
if len(parts) >= 2 and parts[1] in cs.SHELL_SYSTEM_DIRECTORIES:
return f"rm targeting system directory: {resolved_str}"
return f"rm targeting path outside project: {resolved_str}"
return None
def _git_escapes_project(cmd_parts: list[str], project_root: Path) -> tuple[bool, str]:
"""Whether git is pointed at a repository outside the project root.
Git reads the target's config and EXECUTES what it finds -- an
`alias.x = !cmd`, `core.fsmonitor`, `core.pager` -- so any writable
directory becomes arbitrary execution. Verified: a planted
`alias.pwn = "!touch FILE"` ran with rc=0 through
`git --git-dir=SCRATCH/.git pwn`.
Keyed on the resolved TARGET rather than the flag, because pointing git
inside the project is ordinary use: `git -C . diff` and
`git --git-dir x log` stay allowed.
"""
if not cmd_parts or cmd_parts[0] != cs.SHELL_CMD_GIT:
return False, ""
index = 1
while index < len(cmd_parts):
arg = cmd_parts[index]
target = None
if arg in cs.SHELL_GIT_REPO_LOCATION_FLAGS:
if index + 1 < len(cmd_parts):
target = cmd_parts[index + 1]
index += 2
elif "=" in arg and _flag_name(arg) in cs.SHELL_GIT_REPO_LOCATION_FLAGS:
target = arg.split("=", 1)[1]
index += 1
else:
index += 1
continue
if target and (reason := _git_target_escape(target, project_root)):
return True, reason
return False, ""
def _git_target_escape(target: str, project_root: Path) -> str | None:
# Joining onto the root is the platform's own absoluteness test: an
# absolute target replaces the root outright, a relative one lands under
# it. A `startswith("/")` check is POSIX-only: on Windows a rooted,
# drive-less target such as `/x` would resolve on the Python process's
# drive rather than the root's, which is the drive git itself runs on
# (its cwd is the root).
try:
resolved = (project_root / target).resolve()
except (OSError, ValueError):
return f"git pointed at an unresolvable path: {target}"
try:
resolved.relative_to(project_root)
except ValueError:
return (
f"git pointed outside the project at {resolved}, whose config "
"git would execute"
)
return None
def _check_pipeline_patterns(full_command: str) -> str | None:
for pattern, reason in PIPELINE_PATTERNS_COMPILED:
if pattern.search(full_command):
return reason
return None
def _check_segment_patterns(segment: str) -> str | None:
for pattern, reason in SEGMENT_PATTERNS_COMPILED:
if pattern.search(segment):
return reason
return None
def _flag_name(arg: str) -> str:
# `--type=bool` and `--type bool` should compare equal on the flag name.
return arg.split("=", 1)[0]
def _is_git_config_exec_key(key: str) -> bool:
key = key.lower()
if key in cs.SHELL_GIT_CONFIG_EXEC_KEYS:
return True
if any(
key.startswith(prefix)
and key.endswith(suffix)
and len(key) > len(prefix) + len(suffix)
for prefix, suffix in cs.SHELL_GIT_CONFIG_EXEC_KEY_PATTERNS
):
return True
# A suffix rule rather than another name: git names program-valued keys
# by convention, and a list of names cannot cover a key nobody enumerated.
return "." in key and key.endswith(cs.SHELL_GIT_CONFIG_EXEC_KEY_SUFFIXES)
def _git_dash_c_exec_key(cmd_parts: list[str]) -> str | None:
"""Return the shell-executing config key set inline via `git -c key=value`.
`git -c` applies a config key for one command without writing a file, so it
reaches the same executable keys `_git_config_exec_key` guards on the write
path (GHSA-wvxg-744g-6pcg).
"""
if not cmd_parts or cmd_parts[0] != cs.SHELL_CMD_GIT:
return None
index = 1
while index < len(cmd_parts):
arg = cmd_parts[index]
# git's own options end at the subcommand; `-c` after it belongs to the
# subcommand and means something else entirely (`git commit -c <ref>`
# reuses a commit message and executes nothing). Reaching this on a
# value that belongs to a preceding flag would stop the scan early, so
# value-taking global flags consume their argument below.
if not arg.startswith("-"):
return None
key = _git_inline_config_key(cmd_parts, index)
if key and _is_git_config_exec_key(key.split("=", 1)[0]):
return key.split("=", 1)[0]
if (
arg in cs.SHELL_GIT_GLOBAL_VALUE_FLAGS
and arg not in cs.SHELL_GIT_OPTIONAL_ARG_FLAGS
):
index += 2
else:
index += 1
return None
def _git_inline_config_key(cmd_parts: list[str], index: int) -> str | None:
"""The `key=value` a git global option at `index` sets inline, if any.
Covers the separated `-c key=value`, the attached `-ckey=value`, and
`--config-env=key=ENVVAR`.
"""
arg = cmd_parts[index]
if arg in cs.SHELL_GIT_INLINE_CONFIG_FLAGS and index + 1 < len(cmd_parts):
return cmd_parts[index + 1]
if arg.startswith("-c") and len(arg) > 2:
return arg[2:]
if arg.startswith("--config-env="):
return arg[len("--config-env=") :]
return None
def _git_subcommand_index(cmd_parts: list[str]) -> int | None:
"""Index of git's subcommand, stepping over global options.
git's own options precede the subcommand and several consume the token
after them, so the subcommand is neither at a fixed position nor simply
the first non-dash token: `git -C dir config ...` puts `dir` there.
Locating it wrongly has caused three separate bypasses, so every caller
uses this one implementation rather than repeating the rule.
"""
index = 1
while index < len(cmd_parts):
arg = cmd_parts[index]
if not arg.startswith("-"):
return index
if (
arg in cs.SHELL_GIT_GLOBAL_VALUE_FLAGS
and arg not in cs.SHELL_GIT_OPTIONAL_ARG_FLAGS
):
index += 2
else:
index += 1
return None
def _git_config_exec_key(cmd_parts: list[str]) -> str | None:
"""Return the shell-executing git config key this command would write.
`git config` writes to keys like `core.sshCommand` plant a value git later
hands to a shell, so the write itself is RCE on the next git operation
(GHSA-2rr7-8xrw-gmhr). Reading a key is fine, and clearing one (`--unset`)
is how a victim recovers, so those return None.
"""
if len(cmd_parts) < 3 or cmd_parts[0] != cs.SHELL_CMD_GIT:
return None
index = _git_subcommand_index(cmd_parts)
if index is None or cmd_parts[index] != cs.SHELL_GIT_SUBCMD_CONFIG:
return None
args = cmd_parts[index + 1 :]
flags = [_flag_name(arg) for arg in args if arg.startswith("-")]
if any(flag in cs.SHELL_GIT_CONFIG_READ_ACTIONS for flag in flags):
return None
if any(flag in cs.SHELL_GIT_CONFIG_UNSET_FLAGS for flag in flags):
return None
return next(
(
arg
for arg in args
if not arg.startswith("-") and _is_git_config_exec_key(arg)
),
None,
)
def _xargs_short_cluster(arg: str) -> int | None:
"""Tokens consumed by a bundled short-flag cluster, or None if unknown.
`xargs -0pt cat` and `xargs -n1 cat` are ordinary spellings, so treating a
cluster as unrecognisable would block routine use and collapse the
fail-closed rule into blocking everything. Returns 1 when the cluster is
self-contained and 2 when its final flag takes the following token.
"""
for position, letter in enumerate(arg[1:], start=1):
short = f"-{letter}"
if short in cs.SHELL_XARGS_BOOLEAN_FLAGS:
continue
if short in cs.SHELL_XARGS_OPTIONAL_ARG_FLAGS:
# Attached-only: the cluster remainder is its value, if any.
return 1
if short in cs.SHELL_XARGS_VALUE_FLAGS:
# A trailing value flag takes the next token; anything after it in
# the cluster is its value and is consumed with it.
return 2 if position == len(arg) - 1 else 1
return None
return 1
def _xargs_launched_index(cmd_parts: list[str]) -> int | None:
"""Index of the program `xargs` would launch, or None if it launches none.
Returns the position rather than the token so the caller can vet the whole
launched command -- program and arguments -- instead of only its name.
A sentinel value means the scan could not determine the program at all.
"""
if not cmd_parts or cmd_parts[0] != cs.SHELL_CMD_XARGS:
return None
args = cmd_parts[1:]
index = 0
while index < len(args):
arg = args[index]
if not arg.startswith("-"):
return index + 1
if arg == "--":
return index + 2 if index + 1 < len(args) else None
consumed = _xargs_option_width(arg)
if consumed is None:
return -1
index += consumed
return None
def _xargs_option_width(arg: str) -> int | None:
"""Tokens an xargs option consumes, or None if it is unrecognised."""
flag = _flag_name(arg)
if flag in cs.SHELL_XARGS_OPTIONAL_ARG_FLAGS:
return 1
if flag in cs.SHELL_XARGS_VALUE_FLAGS:
if "=" in arg:
return 1
bundled = len(arg) > 2 and not arg.startswith("--")
return 1 if bundled else 2
if flag in cs.SHELL_XARGS_BOOLEAN_FLAGS:
return 1
if len(arg) > 2 and not arg.startswith("--"):
return _xargs_short_cluster(arg)
return None
def _xargs_launched_command(cmd_parts: list[str]) -> str | None:
"""Return the program `xargs` would launch, or None if it launches none.
`_validate_segment` only ever checks a segment's own base command, so an
allowlisted launcher reaches interpreters the allowlist deliberately omits
(GHSA-wvxg-744g-6pcg). Bare `xargs` defaults to `echo` and launches nothing
of its own.
"""
index = _xargs_launched_index(cmd_parts)
if index is None:
return None
if index < 0:
return cs.SHELL_XARGS_UNKNOWN_LAUNCH
return cmd_parts[index]
def _awk_program_skeleton(program: str) -> str:
"""The awk program with string literals blanked out.
A `>` inside a printed string is text, not a redirect, and a filename or
message can contain one. sed's scanner already blanks its user-supplied
spans for the same reason; awk's did not, so `print "a>b"` was refused
while `printf("x") > "f"` slipped past a parenthesis exclusion.
"""
out: list[str] = []
index = 0
while index < len(program):
char = program[index]
if char == '"':
end = _delimited_end(program, index + 1, '"')
if end >= len(program):
return cs.SHELL_AWK_UNPARSEABLE
out.append('""')
index = end + 1
continue
if char == "/" and _awk_regex_starts_here(program, index):
if "/" not in program[index + 1 :]:
# Unterminated: blanking would swallow the rest of the
# program and could hide a construct behind it, so signal
# that the program cannot be scanned.
return cs.SHELL_AWK_UNPARSEABLE
# A /regex/ literal is user text too: `/a|b/` is alternation, not
# a pipe to a command, and gsub(/x|y/,...) is everyday awk.
end = _delimited_end(program, index + 1, "/")
out.append("//")
index = min(end + 1, len(program))
continue
out.append(char)
index += 1
return "".join(out)
def _delimited_end(text: str, start: int, delim: str) -> int:
"""Index of the first unescaped `delim` at or after `start`.
A backslash escapes the character after it. Returns `len(text)` or past it
when the literal is unterminated, so callers compare against the length.
"""
end = start
while end < len(text) and text[end] != delim:
if text[end] == "\\":
end += 1
end += 1
return end
def _awk_regex_starts_here(program: str, index: int) -> bool:
"""Whether the `/` at `index` opens a regex literal rather than division.
awk allows both; a regex can only appear where a value is expected, so it
follows an operator, a delimiter, or the start of the program -- never a
name, number or closing bracket, which is where division follows.
"""
for char in reversed(program[:index]):
if char.isspace():
continue
return not (char.isalnum() or char in "_)]$")
return True
def _awk_redirects_output(program: str) -> bool:
"""Whether an awk program redirects print/printf output to a file.
A redirect's `>` sits at the top level of the output list; a comparison
sits inside parentheses. Tracking depth is awk's actual grammar, where
every spelling-based attempt failed: excluding parentheses hid
`printf("x") > "f"`, and excluding a `$` target hid `print $2 > $1`,
which was verified writing a file.
"""
skeleton = _awk_program_skeleton(program)
for match in re.finditer(r"\b(?:print|printf)\b", skeleton):
depth = 0
index = match.end()
while index < len(skeleton) and skeleton[index] not in ";}":
char = skeleton[index]
if char == "(":
depth += 1
elif char == ")":
depth -= 1
elif char == ">" and depth <= 0:
return True
index += 1
return False
def _awk_exec_construct(cmd_parts: list[str]) -> str | None:
"""Return the construct through which this awk program reaches a command.
awk can only launch a subprocess or write a file through a fixed set of
constructs, so they are detected in the program text itself. Matching the
surrounding spelling was defeated by every indirection tried -- a variable
holding the command name, a -v assignment, string concatenation, a program
supplied with -f, a newline in the program -- because the command name
need never appear literally. The constructs cannot be avoided.
"""
if not cmd_parts or cmd_parts[0] != cs.SHELL_CMD_AWK:
return None
# A program given with -f lives in a file this validator cannot read, so
# its contents are unknowable and it is refused outright.
if _awk_names_program_file(cmd_parts):
return "a program file this validator cannot inspect"
index = 1
while index < len(cmd_parts):
arg = cmd_parts[index]
if arg.startswith("-"):
# A value flag consumes the following token, which is its value
# and not the program; the attached spelling (-vc=id) carries it.
# An OPTIONAL-argument flag takes its value attached only, so the
# next token is still the program and must not be stepped over.
if (
arg in cs.SHELL_AWK_VALUE_FLAGS
and arg not in cs.SHELL_AWK_OPTIONAL_ARG_FLAGS
):
if (
reason := _awk_flag_operand_construct(cmd_parts, index)
) is not None:
return reason
index += 2
else:
index += 1
continue
# Only the first non-flag argument is the program; the rest are files.
return _awk_program_construct(arg)
return None
def _awk_names_program_file(cmd_parts: list[str]) -> bool:
# `-f x`, `-fx` (attached) and `--file=x` all name a program file. Matching
# only the exact token `-f` let the attached spellings through.
return any(
a == "-f"
or (a.startswith("-f") and len(a) > 2)
or _flag_name(a) in cs.SHELL_AWK_PROGRAM_FILE_FLAGS
for a in cmd_parts[1:]
if a.startswith("-")
)
def _awk_flag_operand_construct(cmd_parts: list[str], index: int) -> str | None:
# The operand is the flag's value under gawk, but a program
# under any implementation lacking the flag, so scan BOTH
# readings rather than pick one -- the same rule sed's -i and
# -l need. Picking gawk's alone stepped over the program.
candidate = cmd_parts[index + 1 : index + 2]
if candidate and not candidate[0].startswith("-"):
return _awk_exec_token(candidate[0])
return None
def _awk_program_construct(program: str) -> str | None:
skeleton = _awk_program_skeleton(program)
if skeleton == cs.SHELL_AWK_UNPARSEABLE:
return "a program with an unterminated string or regex"
if _awk_redirects_output(program):
return "redirect to a file"
return _awk_exec_token(skeleton)
def _awk_exec_token(text: str) -> str | None:
for pattern, reason in cs.SHELL_AWK_EXEC_TOKENS:
if re.search(pattern, text):
return reason
return None
def _git_exec_subcommand(cmd_parts: list[str]) -> str | None:
"""Return the git subcommand form that would run a caller-supplied command.
`git` is allowlisted and only its inline-config exec keys were guarded, so
`filter-branch --tree-filter`, `bisect run` and `submodule foreach` reached
a subprocess without meeting the allowlist at all.
"""
if len(cmd_parts) < 3 or cmd_parts[0] != cs.SHELL_CMD_GIT:
return None
subcommand_index = _git_subcommand_index(cmd_parts)
subcommand = cmd_parts[subcommand_index] if subcommand_index is not None else None
if subcommand not in cs.SHELL_GIT_EXEC_SUBCOMMANDS:
return None
for arg in cmd_parts[1:]:
if _flag_name(arg) in cs.SHELL_GIT_EXEC_SUBCOMMAND_ARGS:
return f"{subcommand} {_flag_name(arg)}"
return None
def _git_exec_flag(cmd_parts: list[str]) -> str | None:
"""Return a git flag that names a program git will run.
These are independent of the subcommand: `rebase --exec`, `difftool
--extcmd` and `mergetool --tool` run locally (the first two verified in a
scratch repo), while `--upload-pack`/`--receive-pack`/`--smtp-server`/
`--gpg-sign` name a program run at the other end of a connection.
"""
if not cmd_parts or cmd_parts[0] != cs.SHELL_CMD_GIT:
return None
# `--exec-path=DIR` moves the directory git resolves subcommands in, so
# any non-builtin name execs `DIR/git-<name>`. Verified: a planted
# `git-pwn` ran with rc=0 through `git --exec-path=DIR pwn`. Only the
# VALUED form redirects; bare `--exec-path` prints the path and exits,
# which is why it is filed as optional-arg for arity and must keep its
# pass.
for arg in cmd_parts[1:]:
if arg.startswith(f"{cs.SHELL_GIT_EXEC_PATH_FLAG}="):
return cs.SHELL_GIT_EXEC_PATH_FLAG
if flag := _program_naming_flag(cmd_parts, cs.SHELL_GIT_EXEC_FLAGS):
return flag
return _git_subcommand_exec_option(cmd_parts)
class _GitExecTables(NamedTuple):
"""The program-running option spellings of one git subcommand."""
subcommand: str
long_options: frozenset[str]
short_flags: frozenset[str]
value_flags: frozenset[str]
lookalikes: frozenset[str]
config_long: str | None
config_short: str | None
def _git_exec_tables(subcommand: str) -> _GitExecTables:
return _GitExecTables(
subcommand,
cs.SHELL_GIT_EXEC_LONG_OPTIONS.get(subcommand, frozenset()),
cs.SHELL_GIT_EXEC_SHORT_FLAGS.get(subcommand, frozenset()),
cs.SHELL_GIT_VALUE_SHORT_FLAGS.get(subcommand, frozenset()),
cs.SHELL_GIT_EXEC_OPTION_LOOKALIKES.get(subcommand, frozenset()),
cs.SHELL_GIT_SUBCOMMAND_CONFIG_LONG_OPTIONS.get(subcommand),
cs.SHELL_GIT_SUBCOMMAND_CONFIG_SHORT_FLAGS.get(subcommand),
)
def _git_subcommand_exec_option(cmd_parts: list[str]) -> str | None:
"""A program-running subcommand option in a spelling the exact match misses.
git's parsers accept abbreviated long options (`rebase --ex`), short
letters (`rebase -x`), attached values (`grep -Oprog`) and clustered short
flags (`rebase -ixprog`); the exact-name scan saw none of them.
"""
index = _git_subcommand_index(cmd_parts)
if index is None:
return None
tables = _git_exec_tables(cmd_parts[index])
args = cmd_parts[index + 1 :]
for position, arg in enumerate(args):
# `--` ends the options only when nothing before it can claim it as a
# value: `grep -e -- -Oprog` gives `--` to -e and still parses -Oprog.
# Which options take a value is not fully tabulated, so any option in
# front keeps the scan going; a dash-leading pathspec after such a
# `--` is refused, which is the safe direction.
previous = args[position - 1] if position else tables.subcommand
if arg == "--" and not previous.startswith("-"):
break
following = args[position + 1] if position + 1 < len(args) else ""
if arg.startswith("--"):
hit = _git_long_exec_option(arg, following, tables)
elif arg.startswith("-"):
hit = _git_short_exec_option(arg, following, tables)
else:
hit = None
if hit:
return hit
return None
def _git_long_exec_option(
arg: str, following: str, tables: _GitExecTables
) -> str | None:
name = _flag_name(arg)
if name in tables.lookalikes or len(name) <= 2:
return None
if (
tables.config_long
and len(name) >= cs.SHELL_GIT_CONFIG_OPTION_MIN_ABBREV
and tables.config_long.startswith(name)
):
setting = arg.split("=", 1)[1] if "=" in arg else following
key = _git_exec_config_setting(setting)
return f"{tables.subcommand} {name} {key}" if key else None
# Prefix one way is an abbreviation; the other way is the scripted
# subcommands' `--tool*` case patterns, which take `--toolx=prog` as
# `--tool=prog`.
if any(
option.startswith(name) or name.startswith(option)
for option in tables.long_options
):
return f"{tables.subcommand} {name}"
return None
def _git_short_exec_option(
arg: str, following: str, tables: _GitExecTables
) -> str | None:
for offset, letter in enumerate(arg[1:], start=2):
if letter in tables.short_flags:
return f"{tables.subcommand} -{letter}"
if letter == tables.config_short:
# The rest of the cluster is the key=value, or the next token is
# when the letter ends it: `-qc core.hooksPath=d`.
key = _git_exec_config_setting(arg[offset:] or following)
return f"{tables.subcommand} -{letter} {key}" if key else None
if letter in tables.value_flags:
return None
return None
def _git_exec_config_setting(setting: str) -> str | None:
"""The key of a subcommand-level `key=value` whose value git runs.
`git clone -c core.hooksPath=DIR` writes the key into the new repository
before checkout, so its hooks run during the clone itself (verified): the
same reach as the top-level `git -c` that `_git_dash_c_exec_key` checks.
Harmless keys (`-c core.autocrlf=false`) stay allowed.
"""
key = setting.split("=", 1)[0]
return key if key and _is_git_config_exec_key(key) else None
def _sed_awaits_operand(script: str) -> bool:
"""Whether a sed script ends in a file command still missing its operand.
`sed -ew /tmp/p` splits as ["-ew", "/tmp/p"], so the `w` and its target
live in different argv entries. Only that shape needs the next token
joined; joining always would scan input filenames as script text.
"""
return bool(re.search(r"[wWrR]\s*$", script))
def _sed_script_skeleton(script: str) -> str:
"""The script with s/// and y/// bodies blanked out.
A regex body or replacement can contain any text, including a letter that
the command-anchor patterns read as a command (`s/x/Iw file/` looked like
a `w` write). Blanking those spans leaves the command structure intact
while removing text the user chose, so the anchor sees only positions
where a command can actually appear.
"""
out = []
index = 0
while index < len(script):
char = script[index]
end = _sed_substitution_end(script, index)
if end is not None:
out.append(char)
out.append(" " * (end - index - 1))
index = end
continue
if char == "/":
# A /regex/ address body is user text too, and a letter inside it
# sits immediately before the command position. Blank it, keeping
# the delimiters so the address still anchors the command.
end = _delimited_end(script, index + 1, "/")
if end < len(script):
out.append("/")
out.append(" " * (end - index - 1))
out.append("/")
index = end + 1
continue
out.append(char)
index += 1
return "".join(out)
def _sed_substitution_end(script: str, index: int) -> int | None:
"""End of the s/// or y/// body starting at `index`, or None if none does."""
if script[index] not in "sy" or index + 1 >= len(script):
return None
delim = script[index + 1]
if delim.isalnum() or delim in " \t\n;{}":
return None
end = index + 2
fields = 0
while end < len(script) and fields < 2:
if script[end] == "\\":
end += 2
continue
if script[end] == delim:
fields += 1
end += 1
return end
def _sed_cluster_tail(arg: str) -> str | None:
"""The script-bearing flag a bundled short-flag cluster ends in, if any.
`-ne` is `-n` and `-e`, and the `-e` claims the NEXT token as its script.
Prefix-matching the raw token instead read the cluster's own tail letter
as the script -- `arg[2:]` of `-ne` is the single character `e` -- so the
real script was never collected and a following `-e 'w FILE'` was never
reached. Verified: `sed -ne p -e 'w FILE'` wrote outside the working
directory with rc=0 while the validator saw only `e`.
Returns the trailing flag (`-e`, `-f`, or an optional-arg flag such as
`-i`) when the cluster ends in one and carries no attached value, `""`
when the cluster is script-free, and None when a letter is unrecognised so
the caller can fail closed.
"""
if not arg.startswith("-") or arg.startswith("--") or len(arg) < 3:
return None
for position, letter in enumerate(arg[1:], start=1):
short = f"-{letter}"
if short in ("-e", "-f"):
# A trailing -e/-f takes the next token; anything after it in the
# cluster is its attached value and belongs to it either way.
return short if position == len(arg) - 1 else ""
if short in cs.SHELL_SED_OPTIONAL_ARG_FLAGS:
tail = _sed_optional_flag_tail(short, position == len(arg) - 1)
if tail is not None:
return tail
continue
if short not in cs.SHELL_SED_KNOWN_FLAGS:
return None
return ""
def _sed_optional_flag_tail(short: str, at_end: bool) -> str | None:
"""`_sed_cluster_tail`'s answer at an optional-arg letter, or None to scan on."""
# In TAIL position this behaves like a standalone `-i`: BSD sed
# takes the next token as the suffix, which pushes the script into
# the slot after it. Returning "" here read the cluster as
# script-free, so nothing was collected and the payload two slots
# along was never scanned -- `sed -ni bak 'w FILE' in.txt` wrote
# outside the project root with rc=0 while `sed -n -i bak ...`,
# the same command unbundled, was correctly refused.
if at_end:
return short
if short in ("-l", "--line-length"):
# Mid-cluster `-l` reads two ways and only one of them ends
# the cluster. GNU takes the remainder as its line length;
# BSD treats `-l` as a pure boolean (it sits inside the
# [-EHalnru] cluster in BSD's own synopsis) and keeps
# parsing, so a trailing `i` is a real `-i` that consumes the
# next token as its suffix and puts the script two slots on.
# Stopping here took the GNU half alone and missed the BSD
# reading entirely: `sed -li bak 'w FILE' in.txt` wrote
# outside the project root with rc=0. Keep scanning, which is
# the reading that can still find a script.
return None
# Any other optional-arg flag is attached-only mid-cluster, so the
# remainder is its value.
return ""
def _sed_names_script_file(cmd_parts: list[str]) -> bool:
"""Whether any argument hands sed a script FILE this validator cannot read.
`-f x`, `-fx`, `--file=x` and a bundled `-nf x` all name one. Matching
only tokens that START with `-f` missed every cluster: `sed -nf s.sed in`
ran an arbitrary script file, verified writing its target with rc=0.
"""
for arg in cmd_parts[1:]:
if not arg.startswith("-"):
continue
if arg == "-f" or (arg.startswith("-f") and len(arg) > 2):
return True
if _flag_name(arg) == "--file":
return True
if (