///|
/// Sanitization and CI failure policy.
pub(all) struct Policy {
name : String
keep_sgr : Bool
keep_hyperlinks : Bool
keep_bell : Bool
mark_dangerous : Bool
remove_bidi : Bool
fail_at : Severity
unknown_severity : Severity
max_sequence_chars : Int
} derive(Debug, Eq)
///|
pub fn Policy::strict() -> Policy {
{
name: "strict",
keep_sgr: false,
keep_hyperlinks: false,
keep_bell: false,
mark_dangerous: true,
remove_bidi: true,
fail_at: Dangerous,
unknown_severity: Dangerous,
max_sequence_chars: 4096,
}
}
///|
pub fn Policy::ci() -> Policy {
{
name: "ci",
keep_sgr: true,
keep_hyperlinks: false,
keep_bell: false,
mark_dangerous: false,
remove_bidi: true,
fail_at: Critical,
unknown_severity: Warning,
max_sequence_chars: 16384,
}
}
///|
pub fn Policy::plain() -> Policy {
{
name: "plain",
keep_sgr: false,
keep_hyperlinks: false,
keep_bell: false,
mark_dangerous: false,
remove_bidi: true,
fail_at: Critical,
unknown_severity: Warning,
max_sequence_chars: 16384,
}
}
///|
pub fn Policy::permissive() -> Policy {
{
name: "permissive",
keep_sgr: true,
keep_hyperlinks: true,
keep_bell: false,
mark_dangerous: false,
remove_bidi: true,
fail_at: Critical,
unknown_severity: Info,
max_sequence_chars: 65536,
}
}
///|
pub fn policy_named(name : String) -> Policy? {
match lower_ascii(name) {
"strict" => Some(Policy::strict())
"ci" => Some(Policy::ci())
"plain" => Some(Policy::plain())
"permissive" => Some(Policy::permissive())
_ => None
}
}
///|
fn make_finding(
event : Event,
rule_id : String,
severity : Severity,
risk : RiskKind,
message : String,
decision : Decision,
) -> Finding {
{
rule_id,
severity,
risk,
message,
offset: event.offset,
sequence: visible(event.raw),
decision,
}
}
///|
fn default_decision(policy : Policy, severity : Severity) -> Decision {
if policy.mark_dangerous &&
severity_rank(severity) >= severity_rank(Dangerous) {
Mark
} else {
Drop
}
}
///|
fn classify_c0(event : Event, policy : Policy) -> Finding {
let code = match event.final_byte {
Some(c) => c.to_int()
None => -1
}
if code == 0x07 {
make_finding(
event,
"TP-C0-001",
Warning,
Notification,
"BEL can trigger an audible or visual notification",
if policy.keep_bell {
Keep
} else {
Drop
},
)
} else if code == 0x08 || code == 0x0d {
make_finding(
event,
"TP-C0-002",
Dangerous,
ScreenRewrite,
"backspace or carriage return can visually overwrite earlier log text",
default_decision(policy, Dangerous),
)
} else if code == 0x00 {
make_finding(
event,
"TP-C0-003",
Warning,
UnknownControl,
"NUL is not meaningful in a text log",
Drop,
)
} else {
make_finding(
event,
"TP-C0-099",
policy.unknown_severity,
UnknownControl,
"non-printing C0/C1 control was removed",
Drop,
)
}
}
///|
fn csi_rule(event : Event, policy : Policy, final_byte : Char) -> Finding {
let command = parse_csi(event)
if final_byte == 'm' {
make_finding(
event,
"TP-CSI-001",
Info,
Styling,
"SGR changes text style or color",
if policy.keep_sgr {
Keep
} else {
Drop
},
)
} else if final_byte == 'A' ||
final_byte == 'B' ||
final_byte == 'C' ||
final_byte == 'D' ||
final_byte == 'E' ||
final_byte == 'F' ||
final_byte == 'G' ||
final_byte == 'H' ||
final_byte == 'f' ||
final_byte == 'd' ||
final_byte == 'e' ||
final_byte == '`' {
make_finding(
event,
"TP-CSI-010",
Dangerous,
CursorMovement,
"cursor movement can reorder or overwrite the visible log",
default_decision(policy, Dangerous),
)
} else if final_byte == 'J' || final_byte == 'K' || final_byte == 'X' {
let severity = match command {
Some(cmd) => if csi_param(cmd, 0, 0) == 3 { Critical } else { Dangerous }
None => Dangerous
}
make_finding(
event,
"TP-CSI-011",
severity,
ScreenErase,
"erase command can hide prior output or clear terminal history",
default_decision(policy, severity),
)
} else if final_byte == 'h' || final_byte == 'l' {
make_finding(
event,
"TP-CSI-012",
Dangerous,
ModeChange,
"terminal mode change may alter input, screen, or paste behavior",
default_decision(policy, Dangerous),
)
} else if final_byte == 'n' || final_byte == 'c' {
make_finding(
event,
"TP-CSI-013",
Warning,
TerminalQuery,
"device query can solicit a terminal response",
Drop,
)
} else if final_byte == 's' ||
final_byte == 'u' ||
final_byte == 'S' ||
final_byte == 'T' ||
final_byte == 'r' {
make_finding(
event,
"TP-CSI-014",
Dangerous,
ScreenRewrite,
"saved cursor, scrolling, or region command changes displayed history",
default_decision(policy, Dangerous),
)
} else if final_byte == 't' {
make_finding(
event,
"TP-CSI-015",
Dangerous,
ModeChange,
"window manipulation command is unsafe in untrusted logs",
default_decision(policy, Dangerous),
)
} else {
make_finding(
event,
"TP-CSI-099",
policy.unknown_severity,
UnknownControl,
"unrecognized CSI command was removed",
default_decision(policy, policy.unknown_severity),
)
}
}
///|
fn classify_osc(event : Event, policy : Policy) -> Finding {
guard parse_osc(event) is Some(command) else {
return make_finding(
event,
"TP-OSC-000",
Dangerous,
MalformedSequence,
"malformed OSC command was removed",
default_decision(policy, Dangerous),
)
}
match command.command {
"0" | "1" | "2" =>
make_finding(
event,
"TP-OSC-001",
Dangerous,
TitleChange,
"OSC title command can disguise a terminal window or tab",
default_decision(policy, Dangerous),
)
"8" => {
let target = osc8_target(command.data)
let severity = if is_suspicious_uri(target) {
Critical
} else {
Dangerous
}
make_finding(
event,
"TP-OSC-008",
severity,
Hyperlink,
if severity == Critical {
"OSC 8 contains a high-risk hyperlink scheme"
} else {
"OSC 8 can make visible text link to a different destination"
},
if policy.keep_hyperlinks && severity != Critical {
Keep
} else {
default_decision(policy, severity)
},
)
}
"52" =>
make_finding(
event,
"TP-OSC-052",
Critical,
ClipboardWrite,
"OSC 52 can replace the user's clipboard",
default_decision(policy, Critical),
)
"4" | "10" | "11" | "12" | "104" | "110" | "111" | "112" =>
make_finding(
event,
"TP-OSC-004",
Dangerous,
ModeChange,
"terminal palette mutation can make text misleading or invisible",
default_decision(policy, Dangerous),
)
"7" =>
make_finding(
event,
"TP-OSC-007",
Warning,
WorkingDirectory,
"working-directory metadata was removed from the log",
Drop,
)
"9" | "777" =>
make_finding(
event,
"TP-OSC-009",
Dangerous,
Notification,
"desktop notification command can create deceptive UI",
default_decision(policy, Dangerous),
)
"1337" =>
make_finding(
event,
"TP-OSC-1337",
Critical,
FileTransfer,
"terminal-specific OSC extension may transfer files or inline content",
default_decision(policy, Critical),
)
"133" =>
make_finding(
event,
"TP-OSC-133",
Warning,
ModeChange,
"shell integration marker was removed from untrusted output",
Drop,
)
_ =>
make_finding(
event,
"TP-OSC-099",
policy.unknown_severity,
UnknownControl,
"unrecognized operating-system command was removed",
default_decision(policy, policy.unknown_severity),
)
}
}
///|
fn osc8_target(data : String) -> String {
let out = StringBuilder::new(size_hint=data.length())
let mut in_target = false
for c in data {
if !in_target && c == ';' {
in_target = true
} else if in_target {
out.write_char(c)
}
}
if in_target {
out.to_string()
} else {
data
}
}
///|
pub fn classify(event : Event, policy : Policy) -> Finding? {
if event.kind == Text {
None
} else if event.kind == C0 || event.kind == C1 {
Some(classify_c0(event, policy))
} else if event.kind == Csi {
match event.final_byte {
Some(final_byte) => Some(csi_rule(event, policy, final_byte))
None =>
Some(
make_finding(
event,
"TP-CSI-000",
Dangerous,
MalformedSequence,
"CSI sequence has no final byte",
default_decision(policy, Dangerous),
),
)
}
} else if event.kind == Osc {
Some(classify_osc(event, policy))
} else if event.kind == Dcs ||
event.kind == Sos ||
event.kind == Pm ||
event.kind == Apc {
Some(
make_finding(
event,
"TP-STR-001",
Critical,
DeviceControl,
"terminal control string can carry device-specific commands or data",
default_decision(policy, Critical),
),
)
} else if event.kind == Escape {
let final_byte = match event.final_byte {
Some(c) => c
None => '\u{0}'
}
if final_byte == '7' ||
final_byte == '8' ||
final_byte == 'D' ||
final_byte == 'E' ||
final_byte == 'M' {
Some(
make_finding(
event,
"TP-ESC-001",
Dangerous,
CursorMovement,
"ESC cursor command can rewrite visible output",
default_decision(policy, Dangerous),
),
)
} else if final_byte == 'c' {
Some(
make_finding(
event,
"TP-ESC-002",
Critical,
ModeChange,
"full terminal reset is unsafe in untrusted output",
default_decision(policy, Critical),
),
)
} else {
Some(
make_finding(
event,
"TP-ESC-099",
policy.unknown_severity,
UnknownControl,
"escape command was removed",
default_decision(policy, policy.unknown_severity),
),
)
}
} else {
Some(
make_finding(
event,
"TP-PARSE-001",
Dangerous,
MalformedSequence,
"invalid or truncated terminal sequence was removed",
default_decision(policy, Dangerous),
),
)
}
}