fix(agent): stop identities when a directory changes owner

crates/agent-channel-client/INTENT.md

@@ -9,6 +9,7 @@ This crate is the agent's side of the channel defined in `agent-channel-proto`.
 - The backoff restarts from 1 s only after a stream has stayed authenticated for 60 s, so that a stream that authenticates and then drops at once keeps backing off.
   After such a long-lived stream closes, the first reconnect waits a delay drawn uniformly from `[0, 30 s]`, so that an authority restart doesn't bring every agent back within the same second.
 - A server message that cannot be decoded, or whose `id` is not a UUID, closes the stream as a transient failure without a reply, so the agent never echoes an unchecked `id` and never sends an envelope larger than `CONTRACT.agent-identity.md` §7.1 allows.
+- Text from the authority, a status message or the reason of a push, is logged with control characters, bidirectional overrides and backslashes escaped and is cut to 256 characters, and an `error-code` that is not a lowercase snake_case code of at most 64 bytes is logged as invalid, so that an authority cannot forge log lines or flood the log.
 
 ## Handshake
 

crates/agent-channel-client/src/closure.rs

@@ -7,6 +7,9 @@
 use std::error::Error;
 use std::fmt;
 
+use tonic::Code;
+
+use crate::loggable::loggable;
 use crate::transport::NotGrpc;
 
 /// Why a stream closed or its opening failed ([CONTRACT.agent-identity.md §7.4 "Termination"][contract]).
@@ -16,15 +19,54 @@ use crate::transport::NotGrpc;
 pub struct Closure {
     pub class: ClosureClass,
     /// The gRPC status the authority closed the stream with, if it sent one.
-    pub grpc_code: Option<tonic::Code>,
+    pub grpc_code: Option<Code>,
     /// The `error-code` trailer.
-    pub error_code: Option<String>,
+    pub error_code: Option<ErrorCode>,
     /// The `server-time` trailer.
     pub server_time: Option<String>,
     /// A description of the cause, for logs.
+    ///
+    /// The message of a status the authority sent is escaped and truncated so that it is safe to log.
     pub reason: String,
 }
 
+/// The `error-code` trailer of a status.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum ErrorCode {
+    /// A value in the form of an error code: snake_case ASCII ([CONTRACT.agent-identity.md §1 "Protocol
+    /// conventions"][contract]), starting with a letter, of at most 64 bytes, and not starting with the enrollment token
+    /// prefix `dvaet1`.
+    ///
+    /// [contract]: ../../../docs/agent-identity/CONTRACT.agent-identity.md#1-protocol-conventions
+    Code(String),
+    /// Any other value, which is not kept so that its text never reaches a log.
+    Invalid,
+}
+
+impl ErrorCode {
+    fn parse(value: &str) -> Self {
+        if value.len() <= 64
+            && value.as_bytes().first().is_some_and(u8::is_ascii_lowercase)
+            && value
+                .bytes()
+                .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
+            && !value.starts_with("dvaet1")
+        {
+            Self::Code(value.to_owned())
+        } else {
+            Self::Invalid
+        }
+    }
+
+    /// Returns the code, or `<invalid>` for a value that is not one.
+    pub fn as_str(&self) -> &str {
+        match self {
+            Self::Code(code) => code,
+            Self::Invalid => "<invalid>",
+        }
+    }
+}
+
 /// What a closure means for the identity ([CONTRACT.agent-identity.md §5.4 "Errors"][contract]).
 ///
 /// [contract]: ../../../docs/agent-identity/CONTRACT.agent-identity.md#54-errors
@@ -81,8 +123,9 @@ impl Closure {
 
     /// Classifies the status a stream ended with.
     ///
-    /// A status the authority sent is classified by its `error-code` trailer; a status the client derived from a
-    /// connection failure carries the failure as its source.
+    /// A status the authority sent is classified by its gRPC code with its `error-code` trailer, as exact pairs, and
+    /// any other pair is transient; a status the client derived from a connection failure carries the failure as its
+    /// source.
     pub(crate) fn from_status(status: &tonic::Status) -> Self {
         let mut cause = status.source();
         while let Some(error) = cause {
@@ -94,26 +137,25 @@ impl Closure {
         if let Some(source) = status.source() {
             return Self::outage(error_chain(status.message(), source));
         }
-        let trailer = |name: &str| {
-            status
-                .metadata()
-                .get(name)
-                .and_then(|value| value.to_str().ok())
-                .map(str::to_owned)
-        };
-        let error_code = trailer("error-code");
-        let class = match error_code.as_deref() {
-            Some("identity_revoked") => ClosureClass::Terminal(TerminalCode::IdentityRevoked),
-            Some("identity_unknown") => ClosureClass::Terminal(TerminalCode::IdentityUnknown),
-            Some("certificate_expired") => ClosureClass::CertificateExpired,
+        let metadata = status.metadata();
+        let error_code = metadata
+            .get("error-code")
+            .map(|value| value.to_str().map_or(ErrorCode::Invalid, ErrorCode::parse));
+        let class = match (status.code(), error_code.as_ref().map(ErrorCode::as_str)) {
+            (Code::PermissionDenied, Some("identity_revoked")) => ClosureClass::Terminal(TerminalCode::IdentityRevoked),
+            (Code::Unauthenticated, Some("identity_unknown")) => ClosureClass::Terminal(TerminalCode::IdentityUnknown),
+            (Code::Unauthenticated, Some("certificate_expired")) => ClosureClass::CertificateExpired,
             _ => ClosureClass::Transient(TransientKind::Outage),
         };
         Self {
             class,
             grpc_code: Some(status.code()),
             error_code,
-            server_time: trailer("server-time"),
-            reason: status.message().to_owned(),
+            server_time: metadata
+                .get("server-time")
+                .and_then(|value| value.to_str().ok())
+                .map(str::to_owned),
+            reason: loggable(status.message()),
         }
     }
 }
@@ -129,3 +171,106 @@ pub(crate) fn error_chain(message: &str, source: &(dyn Error + 'static)) -> Stri
     }
     chain
 }
+
+#[cfg(test)]
+mod tests {
+    use rstest::rstest;
+    use tonic::Status;
+    use tonic::metadata::{MetadataMap, MetadataValue};
+
+    use super::*;
+
+    /// A status as the authority sends it, with an `error-code` trailer.
+    fn status(code: Code, error_code: &str) -> Status {
+        let mut metadata = MetadataMap::new();
+        metadata.insert("error-code", error_code.parse().expect("ASCII error code"));
+        Status::with_metadata(code, "closed by the test", metadata)
+    }
+
+    // oracle: contract docs/agent-identity/CONTRACT.agent-identity.md §channel-termination
+    // oracle: contract docs/agent-identity/CONTRACT.agent-identity.md §errors
+    #[rstest]
+    #[case::revoked(
+        Code::PermissionDenied,
+        "identity_revoked",
+        ClosureClass::Terminal(TerminalCode::IdentityRevoked)
+    )]
+    #[case::unknown(
+        Code::Unauthenticated,
+        "identity_unknown",
+        ClosureClass::Terminal(TerminalCode::IdentityUnknown)
+    )]
+    #[case::expired(Code::Unauthenticated, "certificate_expired", ClosureClass::CertificateExpired)]
+    fn classifies_each_status_and_error_code_pair_of_the_contract(
+        #[case] code: Code,
+        #[case] error_code: &str,
+        #[case] class: ClosureClass,
+    ) {
+        assert_eq!(Closure::from_status(&status(code, error_code)).class, class);
+    }
+
+    // oracle: contract docs/agent-identity/CONTRACT.agent-identity.md §channel-termination
+    // oracle: contract docs/agent-identity/CONTRACT.agent-identity.md §errors
+    #[rstest]
+    #[case::revoked_unauthenticated(Code::Unauthenticated, "identity_revoked")]
+    #[case::revoked_ok(Code::Ok, "identity_revoked")]
+    #[case::revoked_unavailable(Code::Unavailable, "identity_revoked")]
+    #[case::unknown_permission_denied(Code::PermissionDenied, "identity_unknown")]
+    #[case::unknown_ok(Code::Ok, "identity_unknown")]
+    #[case::expired_permission_denied(Code::PermissionDenied, "certificate_expired")]
+    #[case::expired_ok(Code::Ok, "certificate_expired")]
+    #[case::expired_internal(Code::Internal, "certificate_expired")]
+    fn a_terminal_or_expired_code_with_another_status_is_an_outage(#[case] code: Code, #[case] error_code: &str) {
+        assert_eq!(
+            Closure::from_status(&status(code, error_code)).class,
+            ClosureClass::Transient(TransientKind::Outage)
+        );
+    }
+
+    // oracle: contract docs/agent-identity/CONTRACT.agent-identity.md §conventions
+    #[rstest]
+    #[case::single_letter("a")]
+    #[case::with_digits("code2_v3")]
+    #[case::future_code("future_code")]
+    // oracle: intent crates/agent-channel-client/INTENT.md §Connection
+    #[case::at_64_bytes("a".repeat(64))]
+    fn keeps_an_error_code_in_snake_case(#[case] error_code: String) {
+        let closure = Closure::from_status(&status(Code::Unauthenticated, &error_code));
+        assert_eq!(closure.error_code, Some(ErrorCode::Code(error_code)));
+    }
+
+    // oracle: contract docs/agent-identity/CONTRACT.agent-identity.md §conventions
+    #[rstest]
+    #[case::empty("")]
+    #[case::upper_case("Identity_Revoked")]
+    #[case::digit_first("2fa")]
+    #[case::underscore_first("_code")]
+    #[case::hyphen("identity-revoked")]
+    #[case::space("identity revoked")]
+    // oracle: intent crates/agent-channel-client/INTENT.md §Connection
+    #[case::past_64_bytes("a".repeat(65))]
+    // oracle: intent docs/agent-identity/INTENT.agent-identity.md §Security invariants
+    #[case::token_prefix("dvaet1_secret")]
+    fn does_not_keep_an_invalid_error_code(#[case] error_code: String) {
+        let closure = Closure::from_status(&status(Code::Unauthenticated, &error_code));
+        assert_eq!(closure.error_code, Some(ErrorCode::Invalid));
+    }
+
+    // oracle: contract docs/agent-identity/CONTRACT.agent-identity.md §conventions
+    #[test]
+    fn does_not_keep_an_error_code_that_is_not_visible_ascii() {
+        let mut metadata = MetadataMap::new();
+        let value = MetadataValue::try_from(b"identity_revoked\xe9".as_slice()).expect("opaque header value");
+        metadata.insert("error-code", value);
+        let closure = Closure::from_status(&Status::with_metadata(Code::PermissionDenied, "", metadata));
+        assert_eq!(closure.error_code, Some(ErrorCode::Invalid));
+        assert_eq!(closure.class, ClosureClass::Transient(TransientKind::Outage));
+    }
+
+    // oracle: intent crates/agent-channel-client/INTENT.md §Connection
+    #[test]
+    fn escapes_the_message_of_a_status_the_authority_sent() {
+        let closure = Closure::from_status(&Status::new(Code::Unavailable, "down\r\nINFO forged"));
+        assert_eq!(closure.reason, r"down\r\nINFO forged");
+    }
+}

crates/agent-channel-client/src/lib.rs

@@ -20,11 +20,12 @@ use crate::transport::Transport;
 
 mod backoff;
 mod closure;
+mod loggable;
 mod stream;
 mod transport;
 
 pub use crate::backoff::{Backoff, uniform_jitter, uniform_spread};
-pub use crate::closure::{Closure, ClosureClass, TerminalCode, TransientKind};
+pub use crate::closure::{Closure, ClosureClass, ErrorCode, TerminalCode, TransientKind};
 
 const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
 const OPENING_TIMEOUT: Duration = Duration::from_secs(30);
@@ -90,8 +91,12 @@ pub enum Event {
     /// The authority answered the `Hello` with `Welcome`: the stream is authenticated.
     Authenticated,
     /// The authority asks the identity to renew; the caller acknowledges `id`.
+    ///
+    /// The authority's `reason` is escaped and truncated so that it is safe to log.
     RenewRequested { id: String, reason: String },
     /// The authority asks for a new stream, which the caller opens before dropping this channel.
+    ///
+    /// The authority's `reason` is escaped and truncated so that it is safe to log.
     Reconnect { id: String, reason: String },
     /// A newer effective config, as a JSON object; the caller acknowledges `id`.
     ConfigUpdate { id: String, config_json: String },

crates/agent-channel-client/src/loggable.rs

@@ -0,0 +1,64 @@
+//! Text the authority chose, made safe to log.
+
+/// The most characters [`loggable`] keeps, escapes included, before its truncation marker.
+const MAX_LOGGED_CHARS: usize = 256;
+/// The marker that ends a truncated text.
+const TRUNCATED: &str = " [truncated]";
+
+/// Returns `text` with each control character, backslash and other character that Rust's `Debug` escapes written as
+/// that escape, cut before the first character that would take it past [`MAX_LOGGED_CHARS`] and then marked as
+/// truncated.
+///
+/// The authority's text can't then start a forged log line or grow a log line without bound.
+pub(crate) fn loggable(text: &str) -> String {
+    let mut logged = String::new();
+    let mut len = 0;
+    for character in text.chars() {
+        let start = logged.len();
+        // Quotes need no escape inside a log field, and escaping them would only clutter the text.
+        if matches!(character, '"' | '\'') {
+            logged.push(character);
+        } else {
+            logged.extend(character.escape_debug());
+        }
+        len += logged[start..].chars().count();
+        if MAX_LOGGED_CHARS < len {
+            logged.truncate(start);
+            logged.push_str(TRUNCATED);
+            break;
+        }
+    }
+    logged
+}
+
+#[cfg(test)]
+mod tests {
+    use rstest::rstest;
+
+    use super::*;
+
+    // oracle: intent crates/agent-channel-client/INTENT.md §Connection
+    #[rstest]
+    #[case::carriage_return_and_line_feed("a\r\nb", r"a\r\nb")]
+    #[case::escape("a\u{1b}[2Jb", r"a\u{1b}[2Jb")]
+    #[case::delete("a\u{7f}b", r"a\u{7f}b")]
+    #[case::c1_next_line("a\u{85}b", r"a\u{85}b")]
+    #[case::nul_and_tab("a\0\tb", r"a\0\tb")]
+    #[case::backslash(r"a\nb", r"a\\nb")]
+    #[case::right_to_left_override("a\u{202e}b", r"a\u{202e}b")]
+    #[case::printable_text_and_quotes("certificate \"rotated\" é", "certificate \"rotated\" é")]
+    fn escapes_each_control_character(#[case] text: &str, #[case] logged: &str) {
+        assert_eq!(loggable(text), logged);
+    }
+
+    // oracle: intent crates/agent-channel-client/INTENT.md §Connection
+    #[rstest]
+    #[case::at_the_bound("a".repeat(256), "a".repeat(256))]
+    #[case::past_the_bound("a".repeat(257), format!("{} [truncated]", "a".repeat(256)))]
+    #[case::multi_byte_character_at_the_bound("é".repeat(257), format!("{} [truncated]", "é".repeat(256)))]
+    #[case::escape_across_the_bound(format!("{}\n", "a".repeat(255)), format!("{} [truncated]", "a".repeat(255)))]
+    #[case::quote_past_the_bound(format!("{}\"", "a".repeat(256)), format!("{} [truncated]", "a".repeat(256)))]
+    fn truncates_past_256_characters_without_splitting_one(#[case] text: String, #[case] logged: String) {
+        assert_eq!(loggable(&text), logged);
+    }
+}