use super::decision::suppression_stage_inner;
use super::detector_policy::DetectorSuppressionPolicy;
use super::path_filter::{
looks_like_raw_base64_file_path, looks_like_secret_scanner_source,
looks_like_vendored_minified_path,
};
use super::shape::{
contains_uuid_v4_substring, looks_like_credential_colliding_punctuation,
looks_like_email_address, looks_like_pure_identifier, looks_like_regex_literal_tail,
looks_like_scheme_prefixed_uri, looks_like_syntactic_punctuation_marker,
looks_like_url_or_path_segment, looks_like_word_separated_identifier,
public_noncredential_shape_with_randomness, PublicShapeScope,
};
use super::token_randomness::{
keep_identifier_gate_with_randomness, keep_word_separated_gate_with_randomness, TokenRandomness,
};
use crate::context;
#[derive(Debug, Clone, Copy)]
pub(crate) struct KnownExampleSuppressionCtx<'a> {
path: Option<&'a str>,
context: context::CodeContext,
source_type: Option<&'a str>,
entropy: Option<f64>,
allow_canonical_hex_key: bool,
allow_base64_blob_shape: bool,
allow_encoded_text_secret: bool,
reject_repeated_blocks: bool,
}
impl<'a> KnownExampleSuppressionCtx<'a> {
pub(crate) fn new(
path: Option<&'a str>,
context: context::CodeContext,
source_type: Option<&'a str>,
) -> Self {
Self {
path,
context,
source_type,
entropy: None,
allow_canonical_hex_key: false,
allow_base64_blob_shape: false,
allow_encoded_text_secret: false,
reject_repeated_blocks: true,
}
}
pub(crate) fn with_entropy(
path: Option<&'a str>,
context: context::CodeContext,
source_type: Option<&'a str>,
entropy: f64,
allow_canonical_hex_key: bool,
allow_base64_blob_shape: bool,
allow_encoded_text_secret: bool,
) -> Self {
Self {
path,
context,
source_type,
entropy: Some(entropy),
allow_canonical_hex_key,
allow_base64_blob_shape,
allow_encoded_text_secret,
reject_repeated_blocks: true,
}
}
pub(crate) fn with_repeated_block_policy(mut self, reject: bool) -> Self {
self.reject_repeated_blocks = reject;
self
}
}
pub(crate) fn suppress_known_example_credential_stage(
credential: &str,
ctx: KnownExampleSuppressionCtx<'_>,
) -> Option<crate::adjudicate::StageId> {
suppression_stage_inner(
credential,
ctx.path,
ctx.context,
ctx.source_type,
false,
false,
ctx.reject_repeated_blocks,
ctx.entropy,
ctx.allow_canonical_hex_key,
ctx.allow_base64_blob_shape,
ctx.allow_encoded_text_secret,
)
}
#[derive(Debug, Clone, Copy)]
pub(crate) struct NamedDetectorSuppressionCtx<'a> {
path: Option<&'a str>,
context: context::CodeContext,
source_type: Option<&'a str>,
detector_rules: Option<&'a DetectorSuppressionPolicy>,
service_anchored: bool,
weak_anchor: bool,
structural_password_slot: bool,
allow_canonical_hex_key_material: bool,
}
impl<'a> NamedDetectorSuppressionCtx<'a> {
pub(crate) fn with_weak_anchor(
path: Option<&'a str>,
context: context::CodeContext,
source_type: Option<&'a str>,
detector_id: &'a str,
service_anchored: bool,
weak_anchor: bool,
structural_password_slot: bool,
) -> Self {
Self::with_weak_anchor_and_key_material_policy(
path,
context,
source_type,
test_detector_suppression_rules(detector_id),
service_anchored,
weak_anchor,
structural_password_slot,
false,
)
}
pub(crate) fn with_weak_anchor_and_key_material_policy(
path: Option<&'a str>,
context: context::CodeContext,
source_type: Option<&'a str>,
detector_rules: Option<&'a DetectorSuppressionPolicy>,
service_anchored: bool,
weak_anchor: bool,
structural_password_slot: bool,
allow_canonical_hex_key_material: bool,
) -> Self {
Self {
path,
context,
source_type,
detector_rules,
service_anchored,
weak_anchor,
structural_password_slot,
allow_canonical_hex_key_material,
}
}
}
pub(crate) fn structural_password_slot_rejection(credential: &str) -> Option<&'static str> {
if credential.bytes().all(|byte| byte.is_ascii_alphabetic())
&& super::token_randomness::is_confident_dictionary_word(credential)
{
Some("dictionary_word_placeholder")
} else if super::token_randomness::has_low_letter_diversity(credential) {
Some("low_letter_diversity_mask")
} else {
None
}
}
pub(crate) fn suppress_named_detector_finding(
credential: &str,
ctx: NamedDetectorSuppressionCtx<'_>,
) -> bool {
suppress_named_detector_finding_stage(credential, ctx).is_some()
}
pub(crate) fn suppress_named_detector_finding_stage(
credential: &str,
ctx: NamedDetectorSuppressionCtx<'_>,
) -> Option<crate::adjudicate::StageId> {
let path = ctx.path;
let context = ctx.context;
let source_type = ctx.source_type;
let detector_rules = ctx.detector_rules;
let service_anchored = ctx.service_anchored;
let weak_anchor = ctx.weak_anchor;
let structural_password_slot = ctx.structural_password_slot;
let randomness = TokenRandomness::for_candidate(credential);
let shape_stage = |reason| Some(crate::adjudicate::StageId::ShapeGate(reason));
if let Some(stage_id) = detector_rules.and_then(|rules| rules.allowlist_stage(path, credential))
{
return Some(stage_id);
}
if crate::placeholder_words::is_exact_entropy_placeholder(credential.as_bytes()) {
return shape_stage("exact_placeholder_value");
}
let apply_tier_b = (!service_anchored && !structural_password_slot) || weak_anchor;
if structural_password_slot {
if let Some(reason) = structural_password_slot_rejection(credential) {
crate::adjudicate::record_example_suppression("pipeline", path, credential, reason);
return shape_stage(reason);
}
}
if apply_tier_b {
if let Some(reason) = super::decision::evasion_decoder_reason(source_type) {
crate::adjudicate::record_example_suppression("pipeline", path, credential, reason);
return shape_stage(reason);
}
}
if apply_tier_b {
if let Some(reason) = public_noncredential_shape_with_randomness(
credential,
PublicShapeScope::WeakAnchor,
&randomness,
) {
crate::adjudicate::record_example_suppression("pipeline", path, credential, reason);
return shape_stage(reason);
}
}
if apply_tier_b
&& keep_identifier_gate_with_randomness(credential, &randomness)
&& looks_like_pure_identifier(credential)
{
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"pure_identifier_no_digit",
);
return shape_stage("pure_identifier_no_digit");
}
if apply_tier_b
&& keep_word_separated_gate_with_randomness(credential, &randomness)
&& looks_like_word_separated_identifier(credential)
{
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"word_separated_identifier",
);
return shape_stage("word_separated_identifier");
}
if apply_tier_b && looks_like_scheme_prefixed_uri(credential) {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"scheme_prefixed_uri",
);
return shape_stage("scheme_prefixed_uri");
}
if looks_like_syntactic_punctuation_marker(credential) {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"syntactic_punctuation_marker",
);
return shape_stage("syntactic_punctuation_marker");
}
if apply_tier_b && looks_like_credential_colliding_punctuation(credential) {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"credential_colliding_punctuation",
);
return shape_stage("credential_colliding_punctuation");
}
if apply_tier_b && looks_like_url_or_path_segment(credential) {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"url_or_path_segment",
);
return shape_stage("url_or_path_segment");
}
if apply_tier_b && contains_uuid_v4_substring(credential) {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"contains_uuid_v4",
);
return shape_stage("contains_uuid_v4");
}
if looks_like_email_address(credential) {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"email_address",
);
return shape_stage("email_address");
}
if looks_like_vendored_minified_path(path) {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"vendored_minified_path",
);
return shape_stage("vendored_minified_path");
}
if source_type.is_some_and(|s| s.contains("binary-strings") || s.contains("archive-binary")) {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"native_binary_strings",
);
return shape_stage("native_binary_strings");
}
if looks_like_secret_scanner_source(path) {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"secret_scanner_source",
);
return shape_stage("secret_scanner_source");
}
if looks_like_raw_base64_file_path(path) && source_type.is_some_and(|s| s == "filesystem") {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"raw_base64_file",
);
return shape_stage("raw_base64_file");
}
if looks_like_regex_literal_tail(credential) {
crate::adjudicate::record_example_suppression(
"pipeline",
path,
credential,
"regex_literal_tail",
);
return shape_stage("regex_literal_tail");
}
if let Some(stage_id) =
detector_rules.and_then(|rules| rules.stopword_stage(path, credential, &randomness))
{
return Some(stage_id);
}
let bypass_shape_gates = (service_anchored || structural_password_slot) && !weak_anchor;
let allow_encoded_text_secret =
!service_anchored && crate::decode_structure::decodes_to_printable_text(credential);
let allow_canonical_hex_key = ctx.allow_canonical_hex_key_material;
suppression_stage_inner(
credential,
path,
context,
source_type,
false,
bypass_shape_gates,
true,
None,
allow_canonical_hex_key,
false,
allow_encoded_text_secret,
)
}
pub(crate) fn detector_weak_anchor(spec: &keyhog_core::DetectorSpec) -> bool {
match detector_weak_anchor_base(spec) {
WeakAnchorBase::Always => true,
WeakAnchorBase::Never => false,
WeakAnchorBase::PerPattern => spec.patterns.iter().any(|pattern| pattern.weak_anchor),
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum WeakAnchorBase {
Always,
Never,
PerPattern,
}
pub(crate) fn detector_weak_anchor_base(spec: &keyhog_core::DetectorSpec) -> WeakAnchorBase {
if spec.weak_anchor {
return WeakAnchorBase::Always;
}
if spec.patterns.iter().any(|pattern| pattern.weak_anchor) {
WeakAnchorBase::PerPattern
} else {
WeakAnchorBase::Never
}
}
#[cfg(test)]
fn test_detector_suppression_rules(
detector_id: &str,
) -> Option<&'static DetectorSuppressionPolicy> {
static RULES: std::sync::LazyLock<DetectorSuppressionPolicy> =
std::sync::LazyLock::new(DetectorSuppressionPolicy::test_fixture);
(detector_id == "test-detector").then_some(&RULES)
}
#[cfg(not(test))]
const fn test_detector_suppression_rules(
_detector_id: &str,
) -> Option<&'static DetectorSuppressionPolicy> {
None
}
#[cfg(test)]
mod weak_anchor_shape_tests {
#[test]
fn test_per_detector_allowlist_and_stopwords() {
use super::{suppress_named_detector_finding_stage, NamedDetectorSuppressionCtx};
use crate::context::CodeContext;
let ctx_allowlisted_path = NamedDetectorSuppressionCtx::with_weak_anchor(
Some("src/allowlisted_path/file.rs"),
CodeContext::Unknown,
Some("filesystem"),
"test-detector",
true,
false,
false,
);
let res = suppress_named_detector_finding_stage("some_secret", ctx_allowlisted_path);
assert_eq!(
res,
Some(crate::adjudicate::StageId::ShapeGate("allowlist_paths"))
);
let ctx_normal_path = NamedDetectorSuppressionCtx::with_weak_anchor(
Some("src/other_path/file.rs"),
CodeContext::Unknown,
Some("filesystem"),
"test-detector",
true,
false,
false,
);
let res = suppress_named_detector_finding_stage("allowlisted_value_12345", ctx_normal_path);
assert_eq!(
res,
Some(crate::adjudicate::StageId::ShapeGate("allowlist_values"))
);
let res =
suppress_named_detector_finding_stage("contains_stopword_here_secret", ctx_normal_path);
assert_eq!(
res,
Some(crate::adjudicate::StageId::ShapeGate("stopwords"))
);
let res =
suppress_named_detector_finding_stage("contains_STOPWORD_HERE_secret", ctx_normal_path);
assert_eq!(
res,
Some(crate::adjudicate::StageId::ShapeGate("stopwords"))
);
let incidental_stopword = "pxidztpv_stopword_here_nxruoapftabufvcsa";
assert!(
crate::suppression::token_randomness::is_random_token(incidental_stopword),
"fixture must exercise the random-token bypass"
);
let res = suppress_named_detector_finding_stage(incidental_stopword, ctx_normal_path);
assert_eq!(
res, None,
"a configured stopword inside a random credential must not suppress it"
);
let res = suppress_named_detector_finding_stage("normal_secret", ctx_normal_path);
assert_eq!(res, None);
}
#[test]
fn exact_placeholder_value_is_suppressed_for_named_detectors() {
use super::{suppress_named_detector_finding_stage, NamedDetectorSuppressionCtx};
use crate::adjudicate::StageId;
use crate::context::CodeContext;
let ctx = NamedDetectorSuppressionCtx::with_weak_anchor(
Some("config/rabbitmq.env"),
CodeContext::Unknown,
Some("filesystem"),
"rabbitmq-management-credentials",
true,
false,
false,
);
let exact = Some(StageId::ShapeGate("exact_placeholder_value"));
for placeholder in [
"password",
"secret",
"default",
"null",
"none",
"undefined",
"empty",
] {
assert_eq!(
suppress_named_detector_finding_stage(placeholder, ctx),
exact,
"named value {placeholder:?} must be suppressed as an exact placeholder",
);
}
assert_eq!(
suppress_named_detector_finding_stage("xK9mP2qR7wZ4nBvT8", ctx),
None,
"a strong random credential must survive the exact-placeholder gate",
);
assert_ne!(
suppress_named_detector_finding_stage("passwordX9mP2qR7wZ4", ctx),
exact,
"a value merely CONTAINING a placeholder word is not an exact match",
);
}
}