Enforce pedantic clippy lints

Author: sl5-guy

.github/workflows/gateway.yml

@@ -41,6 +41,7 @@ jobs:
       - uses: actions/checkout@v5
       - uses: dtolnay/rust-toolchain@stable
       - uses: Swatinem/rust-cache@v2
+      - run: cargo clippy --locked --all-targets
       - run: cargo test --locked
 
   image:

Cargo.toml

@@ -3,6 +3,9 @@ name = "agent_gateway"
 version = "0.1.0"
 edition = "2024"
 
+[lints.clippy]
+pedantic = "deny"
+
 [dependencies]
 tokio = { version = "1", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "time"] }
 rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs", "std", "tls12"] }

src/config.rs

@@ -41,6 +41,12 @@ pub struct PolicyConfig {
 }
 
 impl Config {
+    /// Load, parse, and validate a TOML config file.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the file cannot be read, the TOML cannot be parsed,
+    /// or any config value fails validation.
     pub fn load(path: &Path) -> anyhow::Result<Self> {
         let contents = std::fs::read_to_string(path)?;
         let config: Config = toml::from_str(&contents)?;
@@ -61,6 +67,12 @@ impl Config {
 }
 
 impl PolicyConfig {
+    /// Resolve the configured database URL from the literal value or env var.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if no source is configured, both sources are configured,
+    /// the configured source is empty, or the environment variable cannot be read.
     pub fn database_url(&self) -> anyhow::Result<String> {
         match (&self.database_url, &self.database_url_env) {
             (Some(url), None) => {
@@ -89,18 +101,22 @@ impl PolicyConfig {
         }
     }
 
+    #[must_use]
     pub fn max_connections(&self) -> u32 {
         self.max_connections.unwrap_or(5)
     }
 
+    #[must_use]
     pub fn connect_timeout(&self) -> Duration {
         Duration::from_millis(self.connect_timeout_ms.unwrap_or(5_000))
     }
 
+    #[must_use]
     pub fn pool_acquire_timeout(&self) -> Duration {
         Duration::from_millis(self.pool_acquire_timeout_ms.unwrap_or(1_000))
     }
 
+    #[must_use]
     pub fn query_timeout(&self) -> Duration {
         Duration::from_millis(self.query_timeout_ms.unwrap_or(500))
     }
@@ -119,13 +135,13 @@ impl PolicyConfig {
 
         match (&self.database_url, &self.database_url_env) {
             (Some(url), None) => {
-                anyhow::ensure!(!url.is_empty(), "policy.database_url must not be empty")
+                anyhow::ensure!(!url.is_empty(), "policy.database_url must not be empty");
             }
             (None, Some(env_name)) => {
                 anyhow::ensure!(
                     !env_name.is_empty(),
                     "policy.database_url_env must not be empty"
-                )
+                );
             }
             (None, None) => {
                 anyhow::bail!("policy.database_url or policy.database_url_env is required")

src/observability.rs

@@ -12,6 +12,12 @@ use crate::config::ObservabilityConfig;
 
 static TRACER_PROVIDER: OnceLock<SdkTracerProvider> = OnceLock::new();
 
+/// Initialize tracing and optional OpenTelemetry export.
+///
+/// # Errors
+///
+/// Returns an error if the OTLP exporter cannot be built or the global tracing
+/// subscriber cannot be initialized.
 pub fn init(config: &ObservabilityConfig) -> anyhow::Result<()> {
     global::set_text_map_propagator(TraceContextPropagator::new());
 
@@ -55,14 +61,12 @@ pub fn init(config: &ObservabilityConfig) -> anyhow::Result<()> {
 }
 
 fn stdout_logging_enabled() -> bool {
-    std::env::var("AGENT_GATEWAY_LOG_STDOUT")
-        .map(|value| {
-            !matches!(
-                value.trim().to_ascii_lowercase().as_str(),
-                "0" | "false" | "off" | "no"
-            )
-        })
-        .unwrap_or(true)
+    std::env::var("AGENT_GATEWAY_LOG_STDOUT").map_or(true, |value| {
+        !matches!(
+            value.trim().to_ascii_lowercase().as_str(),
+            "0" | "false" | "off" | "no"
+        )
+    })
 }
 
 pub fn shutdown() {

src/policy.rs

@@ -50,6 +50,13 @@ impl PostgresPolicyEngine {
     }
 }
 
+/// Build the configured policy engine.
+///
+/// # Errors
+///
+/// Returns an error if the database URL is invalid, the database cannot be
+/// reached, the schema version does not match, or the client extension OID is
+/// invalid.
 pub async fn build_engine(config: &PolicyConfig) -> anyhow::Result<Arc<dyn PolicyEngine>> {
     let db_pool = build_pg_pool(config).await?;
     RegistryStore::verify_schema_version(&db_pool).await?;
@@ -284,7 +291,7 @@ fn certificate_subject(
                 return Err("extension value contains trailing bytes".into());
             }
             Ok(CertificateSubject {
-                identity: v.string().to_owned(),
+                identity: v.string().clone(),
                 public_key_spki_der,
             })
         }

src/proxy.rs

@@ -32,7 +32,7 @@ impl Extractor for HeaderExtractor<'_> {
     }
 
     fn keys(&self) -> Vec<&str> {
-        self.0.keys().map(|name| name.as_str()).collect()
+        self.0.keys().map(http::HeaderName::as_str).collect()
     }
 }
 
@@ -96,37 +96,22 @@ impl ProxyService {
                     source_identity
                 }
                 PolicyDecision::Deny {
-                    source_identity: Some(source_identity),
+                    source_identity,
                     reason,
                 } => {
-                    warn!(
-                        source_identity = %source_identity,
-                        source_peer_addr = %self.source_peer_addr,
-                        dest_authority = %dest.authority,
-                        policy_decision = "deny",
-                        deny_reason = %reason,
-                        "CONNECT denied"
-                    );
-                    return response(StatusCode::FORBIDDEN, "forbidden");
-                }
-                PolicyDecision::Deny {
-                    source_identity: None,
-                    reason,
-                } => {
-                    warn!(
-                        source_peer_addr = %self.source_peer_addr,
-                        dest_authority = %dest.authority,
-                        policy_decision = "deny",
-                        deny_reason = %reason,
-                        "CONNECT denied"
+                    log_denial(
+                        source_identity.as_deref(),
+                        self.source_peer_addr,
+                        &dest.authority,
+                        &reason,
                     );
                     return response(StatusCode::FORBIDDEN, "forbidden");
                 }
             };
 
             // Connect to destination BEFORE returning 200 so the client knows
             // the tunnel is actually established.
-            let mut upstream = match TcpStream::connect((&*dest.host, dest.port)).await {
+            let upstream = match TcpStream::connect((&*dest.host, dest.port)).await {
                 Ok(s) => s,
                 Err(e) => {
                     error!(
@@ -141,50 +126,12 @@ impl ProxyService {
             };
 
             let on_upgrade = hyper::upgrade::on(req);
-            let source_peer_addr = self.source_peer_addr;
-
-            let tunnel_span = tracing::Span::current();
-            tokio::spawn(
-                async move {
-                    let upgraded = match on_upgrade.await {
-                        Ok(u) => u,
-                        Err(e) => {
-                            warn!(
-                                source_identity = %source_identity,
-                                source_peer_addr = %source_peer_addr,
-                                dest_authority = %dest.authority,
-                                error = %e,
-                                "upgrade failed"
-                            );
-                            return;
-                        }
-                    };
-
-                    let mut downstream = hyper_util::rt::TokioIo::new(upgraded);
-
-                    match copy_bidirectional(&mut downstream, &mut upstream).await {
-                        Ok((up, down)) => {
-                            info!(
-                                source_identity = %source_identity,
-                                source_peer_addr = %source_peer_addr,
-                                dest_authority = %dest.authority,
-                                bytes_client_to_dest = up,
-                                bytes_dest_to_client = down,
-                                "tunnel closed"
-                            );
-                        }
-                        Err(e) => {
-                            error!(
-                                source_identity = %source_identity,
-                                source_peer_addr = %source_peer_addr,
-                                dest_authority = %dest.authority,
-                                error = %e,
-                                "tunnel error"
-                            );
-                        }
-                    }
-                }
-                .instrument(tunnel_span),
+            spawn_tunnel(
+                on_upgrade,
+                upstream,
+                source_identity,
+                self.source_peer_addr,
+                dest.authority,
             );
 
             response(StatusCode::OK, "")
@@ -214,6 +161,7 @@ impl MakeProxyService {
         Self { policy_engine }
     }
 
+    #[must_use]
     pub fn make_service(
         &self,
         peer_certs: Vec<CertificateDer<'static>>,
@@ -223,12 +171,91 @@ impl MakeProxyService {
     }
 }
 
+#[must_use]
 pub fn extract_peer_certs(conn: &ServerConnection) -> Vec<CertificateDer<'static>> {
     conn.peer_certificates()
-        .map(|certs| certs.to_vec())
+        .map(<[CertificateDer<'_>]>::to_vec)
         .unwrap_or_default()
 }
 
+fn log_denial(
+    source_identity: Option<&str>,
+    source_peer_addr: SocketAddr,
+    dest_authority: &str,
+    reason: &str,
+) {
+    if let Some(source_identity) = source_identity {
+        warn!(
+            source_identity = %source_identity,
+            source_peer_addr = %source_peer_addr,
+            dest_authority = %dest_authority,
+            policy_decision = "deny",
+            deny_reason = %reason,
+            "CONNECT denied"
+        );
+    } else {
+        warn!(
+            source_peer_addr = %source_peer_addr,
+            dest_authority = %dest_authority,
+            policy_decision = "deny",
+            deny_reason = %reason,
+            "CONNECT denied"
+        );
+    }
+}
+
+fn spawn_tunnel(
+    on_upgrade: hyper::upgrade::OnUpgrade,
+    mut upstream: TcpStream,
+    source_identity: String,
+    source_peer_addr: SocketAddr,
+    dest_authority: String,
+) {
+    let tunnel_span = tracing::Span::current();
+    tokio::spawn(
+        async move {
+            let upgraded = match on_upgrade.await {
+                Ok(u) => u,
+                Err(e) => {
+                    warn!(
+                        source_identity = %source_identity,
+                        source_peer_addr = %source_peer_addr,
+                        dest_authority = %dest_authority,
+                        error = %e,
+                        "upgrade failed"
+                    );
+                    return;
+                }
+            };
+
+            let mut downstream = hyper_util::rt::TokioIo::new(upgraded);
+
+            match copy_bidirectional(&mut downstream, &mut upstream).await {
+                Ok((up, down)) => {
+                    info!(
+                        source_identity = %source_identity,
+                        source_peer_addr = %source_peer_addr,
+                        dest_authority = %dest_authority,
+                        bytes_client_to_dest = up,
+                        bytes_dest_to_client = down,
+                        "tunnel closed"
+                    );
+                }
+                Err(e) => {
+                    error!(
+                        source_identity = %source_identity,
+                        source_peer_addr = %source_peer_addr,
+                        dest_authority = %dest_authority,
+                        error = %e,
+                        "tunnel error"
+                    );
+                }
+            }
+        }
+        .instrument(tunnel_span),
+    );
+}
+
 fn response(status: StatusCode, message: &str) -> Response<ProxyBody> {
     let body: ProxyBody = if message.is_empty() {
         Empty::<Bytes>::new().boxed()

src/registry.rs

@@ -60,7 +60,7 @@ impl RegistryStore {
         subject_public_key_spki_der: &[u8],
     ) -> anyhow::Result<Vec<CandidatePermission>> {
         let query = sqlx::query_as::<_, CandidatePermission>(
-            r#"
+            r"
             SELECT
                 p.permission_id,
                 p.subject_identity,
@@ -90,7 +90,7 @@ impl RegistryStore {
               AND p.not_after > now()
             ORDER BY p.not_after DESC
             LIMIT 16
-            "#,
+            ",
         )
         .bind(subject_identity)
         .bind(destination)
@@ -110,7 +110,7 @@ impl RegistryStore {
         permission_not_after: DateTime<Utc>,
     ) -> anyhow::Result<bool> {
         let query = sqlx::query_scalar::<_, bool>(
-            r#"
+            r"
             SELECT EXISTS (
                 SELECT 1
                 FROM principal_key_permissions
@@ -122,7 +122,7 @@ impl RegistryStore {
                   AND not_before <= $3
                   AND not_after >= $4
             )
-            "#,
+            ",
         )
         .bind(signing_key_id)
         .bind(destination)