krabka_verified/
schema.rs

1//! Schema-registry failure classification and fail-open admission.
2
3use creusot_std::prelude::*;
4
5/// Security-relevant class of a failed schema-registry lookup.
6#[krabka_macros::model_type]
7pub enum SchemaFailureKind {
8    /// The registry answered that the schema ID does not exist.
9    Unknown,
10    /// No authoritative answer is available yet: transport, throttling, or 5xx.
11    Transient,
12    /// The registry definitively rejected the request, such as with another 4xx.
13    Permanent,
14    /// A successful response could not be decoded into the required shape.
15    Malformed,
16}
17
18/// Whether a failed lookup rejects the record or admits it without validation.
19#[krabka_macros::model_type]
20pub enum SchemaFailureDecision {
21    Reject,
22    AllowUnvalidated,
23}
24
25/// Apply the configured fail-open policy to a classified registry failure.
26///
27/// Only a transient failure may be admitted, and only when the operator opted
28/// into fail-open behavior. Definite and malformed answers always fail closed.
29#[ensures((result == SchemaFailureDecision::AllowUnvalidated)
30    == (fail_open && failure == SchemaFailureKind::Transient))]
31#[must_use]
32pub fn schema_failure_decision(
33    fail_open: bool,
34    failure: SchemaFailureKind,
35) -> SchemaFailureDecision {
36    match failure {
37        SchemaFailureKind::Transient if fail_open => SchemaFailureDecision::AllowUnvalidated,
38        _ => SchemaFailureDecision::Reject,
39    }
40}
41
42/// A schema-validated record field's role in subject selection.
43#[krabka_macros::model_type]
44pub enum SchemaFieldRole {
45    Key,
46    Value,
47}
48
49/// Whether the schema gate skips a field or validates it under one role.
50#[krabka_macros::model_type]
51pub enum SchemaFieldAction {
52    Skip,
53    CheckKey,
54    CheckValue,
55}
56
57/// Whether every applicable field in a record batch was admitted.
58#[krabka_macros::model_type]
59pub enum SchemaBatchAdmission {
60    Admit,
61    Reject,
62}
63
64/// Decode the exact big-endian schema ID from a complete Confluent prefix.
65///
66/// The adapter supplies the five prefix bytes only after establishing that
67/// they exist. Magic zero is the sole admitted framing version.
68#[ensures(match result {
69    None => magic != 0u8,
70    Some(id) => magic == 0u8
71        && id@ == id_0@ * 16_777_216 + id_1@ * 65_536 + id_2@ * 256 + id_3@,
72})]
73#[must_use]
74pub fn schema_frame_id(magic: u8, id_0: u8, id_1: u8, id_2: u8, id_3: u8) -> Option<u32> {
75    if magic != 0 {
76        return None;
77    }
78    Some(
79        u32::from(id_0) * 16_777_216
80            + u32::from(id_1) * 65_536
81            + u32::from(id_2) * 256
82            + u32::from(id_3),
83    )
84}
85
86/// Select exactly the configured, non-null key or value field.
87#[ensures((result == SchemaFieldAction::CheckKey)
88    == (key_enabled && present && role == SchemaFieldRole::Key))]
89#[ensures((result == SchemaFieldAction::CheckValue)
90    == (value_enabled && present && role == SchemaFieldRole::Value))]
91#[ensures((result == SchemaFieldAction::Skip)
92    == (!present
93        || (role == SchemaFieldRole::Key && !key_enabled)
94        || (role == SchemaFieldRole::Value && !value_enabled)))]
95#[must_use]
96pub fn schema_field_action(
97    key_enabled: bool,
98    value_enabled: bool,
99    role: SchemaFieldRole,
100    present: bool,
101) -> SchemaFieldAction {
102    if present {
103        match role {
104            SchemaFieldRole::Key if key_enabled => SchemaFieldAction::CheckKey,
105            SchemaFieldRole::Value if value_enabled => SchemaFieldAction::CheckValue,
106            SchemaFieldRole::Key | SchemaFieldRole::Value => SchemaFieldAction::Skip,
107        }
108    } else {
109        SchemaFieldAction::Skip
110    }
111}
112
113/// Admit a batch only after a complete walk where every applicable field was admitted.
114#[ensures((result == SchemaBatchAdmission::Admit)
115    == (walk_complete && applicable == admitted))]
116#[must_use]
117pub fn schema_batch_admission(
118    walk_complete: bool,
119    applicable: u64,
120    admitted: u64,
121) -> SchemaBatchAdmission {
122    if walk_complete && applicable == admitted {
123        SchemaBatchAdmission::Admit
124    } else {
125        SchemaBatchAdmission::Reject
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use {super::*, assert2::check};
132
133    #[yare::parameterized(
134        unknown = { SchemaFailureKind::Unknown, SchemaFailureDecision::Reject },
135        transient = { SchemaFailureKind::Transient, SchemaFailureDecision::AllowUnvalidated },
136        permanent = { SchemaFailureKind::Permanent, SchemaFailureDecision::Reject },
137        malformed = { SchemaFailureKind::Malformed, SchemaFailureDecision::Reject },
138    )]
139    fn fail_open_admits_only_transient_failures(
140        failure: SchemaFailureKind,
141        open_decision: SchemaFailureDecision,
142    ) {
143        check!(schema_failure_decision(false, failure) == SchemaFailureDecision::Reject);
144        check!(schema_failure_decision(true, failure) == open_decision);
145    }
146
147    #[yare::parameterized(
148        ordinary = { 0, 0x01, 0x23, 0x45, 0x67, Some(0x0123_4567) },
149        one = { 0, 0, 0, 0, 1, Some(1) },
150        max_id = { 0, 0xff, 0xff, 0xff, 0xff, Some(u32::MAX) },
151        wrong_magic = { 1, 0x01, 0x23, 0x45, 0x67, None },
152        invalid_magic = { 0xff, 0, 0, 0, 1, None },
153    )]
154    fn confluent_framing_is_magic_zero_then_a_big_endian_id(
155        magic: u8,
156        id_0: u8,
157        id_1: u8,
158        id_2: u8,
159        id_3: u8,
160        expected: Option<u32>,
161    ) {
162        check!(schema_frame_id(magic, id_0, id_1, id_2, id_3) == expected);
163    }
164
165    #[yare::parameterized(
166        key_enabled = {
167            "key validation checks a key",
168            true,
169            false,
170            SchemaFieldRole::Key,
171            true,
172            SchemaFieldAction::CheckKey,
173        },
174        value_enabled = {
175            "value validation checks a value",
176            false,
177            true,
178            SchemaFieldRole::Value,
179            true,
180            SchemaFieldAction::CheckValue,
181        },
182        both_key = {
183            "both enabled checks a key",
184            true,
185            true,
186            SchemaFieldRole::Key,
187            true,
188            SchemaFieldAction::CheckKey,
189        },
190        both_value = {
191            "both enabled checks a value",
192            true,
193            true,
194            SchemaFieldRole::Value,
195            true,
196            SchemaFieldAction::CheckValue,
197        },
198        null_key = {
199            "a null key is exempt",
200            true,
201            true,
202            SchemaFieldRole::Key,
203            false,
204            SchemaFieldAction::Skip,
205        },
206        null_value = {
207            "a null value (tombstone) is exempt",
208            true,
209            true,
210            SchemaFieldRole::Value,
211            false,
212            SchemaFieldAction::Skip,
213        },
214        value_only_skips_key = {
215            "value-only validation skips the key",
216            false,
217            true,
218            SchemaFieldRole::Key,
219            true,
220            SchemaFieldAction::Skip,
221        },
222        key_only_skips_value = {
223            "key-only validation skips the value",
224            true,
225            false,
226            SchemaFieldRole::Value,
227            true,
228            SchemaFieldAction::Skip,
229        },
230        disabled = {
231            "validation disabled skips everything",
232            false,
233            false,
234            SchemaFieldRole::Value,
235            true,
236            SchemaFieldAction::Skip,
237        },
238    )]
239    fn only_configured_non_null_fields_are_validated(
240        scenario: &str,
241        key_enabled: bool,
242        value_enabled: bool,
243        role: SchemaFieldRole,
244        present: bool,
245        expected: SchemaFieldAction,
246    ) {
247        check!(
248            schema_field_action(key_enabled, value_enabled, role, present) == expected,
249            "{scenario}"
250        );
251    }
252
253    #[yare::parameterized(
254        empty = { true, 0, 0, SchemaBatchAdmission::Admit },
255        all_admitted = { true, 4, 4, SchemaBatchAdmission::Admit },
256        one_missing = { true, 4, 3, SchemaBatchAdmission::Reject },
257        too_many_admitted = { true, 3, 4, SchemaBatchAdmission::Reject },
258        incomplete = { false, 4, 4, SchemaBatchAdmission::Reject },
259        incomplete_empty = { false, 0, 0, SchemaBatchAdmission::Reject },
260    )]
261    fn batch_admits_only_a_complete_walk_with_every_field_admitted(
262        walk_complete: bool,
263        applicable: u64,
264        admitted: u64,
265        expected: SchemaBatchAdmission,
266    ) {
267        check!(schema_batch_admission(walk_complete, applicable, admitted) == expected);
268    }
269}
krabka_verified/schema.rs

krabka_verified/
schema.rs
