Add YAML-configured ACLs with Basic and IP authorization

src/acl.rs

@@ -0,0 +1,550 @@
+use std::{collections::BTreeSet, net::IpAddr, sync::Arc};
+
+use argon2::{
+    Algorithm, Argon2, Params, Version,
+    password_hash::{PasswordHasher, PasswordVerifier, phc::PasswordHash},
+};
+use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
+use http::{HeaderMap, header::PROXY_AUTHORIZATION};
+use ipnet::IpNet;
+use thiserror::Error;
+use tokio::sync::Semaphore;
+use zeroize::Zeroizing;
+
+use crate::{
+    BoxError,
+    config::{AclConfig, Action, RuleConfig},
+};
+
+const MAX_AUTHORIZATION_HEADER_LEN: usize = 4096;
+const MAX_USERNAME_LEN: usize = 256;
+const MAX_PASSWORD_LEN: usize = 1024;
+const MAX_CONCURRENT_VERIFICATIONS: usize = 4;
+const MIN_ARGON2_MEMORY_KIB: u32 = 4096;
+const MAX_ARGON2_MEMORY_KIB: u32 = 65_536;
+const MAX_ARGON2_ITERATIONS: u32 = 10;
+const MAX_ARGON2_PARALLELISM: u32 = 4;
+
+pub(crate) struct Acl {
+    default_action: Action,
+    users: std::collections::BTreeMap<String, User>,
+    rules: Vec<Rule>,
+    has_user_rules: bool,
+    verification_limit: Arc<Semaphore>,
+}
+
+struct User {
+    password_hash: String,
+}
+
+struct Rule {
+    name: String,
+    action: Action,
+    users: BTreeSet<String>,
+    networks: Vec<IpNet>,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum Decision {
+    Allow,
+    Deny,
+    Challenge,
+}
+
+impl Decision {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Allow => "allow",
+            Self::Deny => "deny",
+            Self::Challenge => "challenge",
+        }
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct AclOutcome {
+    pub decision: Decision,
+    pub reason: &'static str,
+    pub rule: Option<String>,
+}
+
+impl AclOutcome {
+    fn rule(rule: &Rule) -> Self {
+        let reason = match (rule.users.is_empty(), rule.networks.is_empty()) {
+            (false, false) => "user_and_network",
+            (false, true) => "user",
+            (true, false) => "network",
+            (true, true) => unreachable!("empty ACL rules are rejected"),
+        };
+        Self {
+            decision: match rule.action {
+                Action::Allow => Decision::Allow,
+                Action::Deny => Decision::Deny,
+            },
+            reason,
+            rule: Some(rule.name.clone()),
+        }
+    }
+
+    fn default(action: Action) -> Self {
+        Self {
+            decision: match action {
+                Action::Allow => Decision::Allow,
+                Action::Deny => Decision::Deny,
+            },
+            reason: "default",
+            rule: None,
+        }
+    }
+
+    fn challenge(reason: &'static str) -> Self {
+        Self {
+            decision: Decision::Challenge,
+            reason,
+            rule: None,
+        }
+    }
+}
+
+impl Acl {
+    pub fn from_config(config: AclConfig) -> Result<Self, AclError> {
+        let mut users = std::collections::BTreeMap::new();
+        for (username, user) in config.users {
+            validate_password_hash(&username, &user.password_hash)?;
+            users.insert(
+                username,
+                User {
+                    password_hash: user.password_hash,
+                },
+            );
+        }
+
+        let mut names = BTreeSet::new();
+        let mut rules = Vec::with_capacity(config.rules.len());
+        for rule in config.rules {
+            let compiled = compile_rule(rule, &users)?;
+            if !names.insert(compiled.name.clone()) {
+                return Err(AclError::DuplicateRule(compiled.name));
+            }
+            rules.push(compiled);
+        }
+
+        let has_user_rules = rules.iter().any(|rule| !rule.users.is_empty());
+        Ok(Self {
+            default_action: config.default_action,
+            users,
+            rules,
+            has_user_rules,
+            verification_limit: Arc::new(Semaphore::new(MAX_CONCURRENT_VERIFICATIONS)),
+        })
+    }
+
+    pub async fn authorize(
+        &self,
+        peer: IpAddr,
+        headers: &HeaderMap,
+    ) -> Result<AclOutcome, AclError> {
+        let credentials = match parse_basic_credentials(headers) {
+            Credentials::Missing => None,
+            Credentials::Invalid => return Ok(AclOutcome::challenge("invalid_credentials")),
+            Credentials::Present(credentials) => Some(credentials),
+        };
+
+        let authenticated_user = match credentials {
+            Some(credentials) => {
+                let Some(user) = self.users.get(&credentials.username) else {
+                    return Ok(AclOutcome::challenge("invalid_credentials"));
+                };
+                if !self
+                    .verify_password(user.password_hash.clone(), credentials.password)
+                    .await?
+                {
+                    return Ok(AclOutcome::challenge("invalid_credentials"));
+                }
+                Some(credentials.username)
+            }
+            None => None,
+        };
+
+        for rule in &self.rules {
+            if rule_matches(rule, peer, authenticated_user.as_deref()) {
+                return Ok(AclOutcome::rule(rule));
+            }
+        }
+
+        if authenticated_user.is_none()
+            && self.default_action == Action::Deny
+            && self.has_user_rules
+        {
+            return Ok(AclOutcome::challenge("authentication_required"));
+        }
+
+        Ok(AclOutcome::default(self.default_action))
+    }
+
+    async fn verify_password(
+        &self,
+        password_hash: String,
+        password: Zeroizing<Vec<u8>>,
+    ) -> Result<bool, AclError> {
+        let permit = self
+            .verification_limit
+            .clone()
+            .acquire_owned()
+            .await
+            .map_err(|_| AclError::AuthenticationTask)?;
+
+        tokio::task::spawn_blocking(move || {
+            let _permit = permit;
+            let parsed =
+                PasswordHash::new(&password_hash).map_err(|_| AclError::AuthenticationTask)?;
+            Ok(Argon2::default()
+                .verify_password(password.as_slice(), &parsed)
+                .is_ok())
+        })
+        .await
+        .map_err(|_| AclError::AuthenticationTask)?
+    }
+}
+
+fn compile_rule(
+    rule: RuleConfig,
+    users: &std::collections::BTreeMap<String, User>,
+) -> Result<Rule, AclError> {
+    if rule.name.trim().is_empty() {
+        return Err(AclError::EmptyRuleName);
+    }
+    if rule.users.is_empty() && rule.networks.is_empty() {
+        return Err(AclError::EmptyRule(rule.name));
+    }
+
+    for username in &rule.users {
+        if !users.contains_key(username) {
+            return Err(AclError::UnknownUser {
+                rule: rule.name,
+                username: username.clone(),
+            });
+        }
+    }
+
+    let mut networks = Vec::with_capacity(rule.networks.len());
+    for network in rule.networks {
+        let parsed = network.parse().map_err(|_| AclError::InvalidNetwork {
+            rule: rule.name.clone(),
+            network,
+        })?;
+        networks.push(parsed);
+    }
+
+    Ok(Rule {
+        name: rule.name,
+        action: rule.action,
+        users: rule.users.into_iter().collect(),
+        networks,
+    })
+}
+
+fn validate_password_hash(username: &str, encoded: &str) -> Result<(), AclError> {
+    let hash = PasswordHash::new(encoded)
+        .map_err(|_| AclError::InvalidPasswordHash(username.to_owned()))?;
+    if hash.algorithm.as_str() != Algorithm::Argon2id.as_str()
+        || hash.version != Some(u32::from(Version::V0x13 as u8))
+    {
+        return Err(AclError::InvalidPasswordHash(username.to_owned()));
+    }
+
+    let params =
+        Params::try_from(&hash).map_err(|_| AclError::InvalidPasswordHash(username.to_owned()))?;
+    if !(MIN_ARGON2_MEMORY_KIB..=MAX_ARGON2_MEMORY_KIB).contains(&params.m_cost())
+        || params.t_cost() == 0
+        || params.t_cost() > MAX_ARGON2_ITERATIONS
+        || params.p_cost() == 0
+        || params.p_cost() > MAX_ARGON2_PARALLELISM
+    {
+        return Err(AclError::UnsafePasswordParameters(username.to_owned()));
+    }
+
+    Ok(())
+}
+
+fn rule_matches(rule: &Rule, peer: IpAddr, authenticated_user: Option<&str>) -> bool {
+    let user_matches = rule.users.is_empty()
+        || authenticated_user.is_some_and(|username| rule.users.contains(username));
+    let network_matches =
+        rule.networks.is_empty() || rule.networks.iter().any(|network| network.contains(&peer));
+    user_matches && network_matches
+}
+
+enum Credentials {
+    Missing,
+    Invalid,
+    Present(BasicCredentials),
+}
+
+struct BasicCredentials {
+    username: String,
+    password: Zeroizing<Vec<u8>>,
+}
+
+fn parse_basic_credentials(headers: &HeaderMap) -> Credentials {
+    let mut values = headers.get_all(PROXY_AUTHORIZATION).iter();
+    let Some(value) = values.next() else {
+        return Credentials::Missing;
+    };
+    if values.next().is_some() || value.len() > MAX_AUTHORIZATION_HEADER_LEN {
+        return Credentials::Invalid;
+    }
+
+    let Ok(value) = value.to_str() else {
+        return Credentials::Invalid;
+    };
+    let mut parts = value.split_ascii_whitespace();
+    let (Some(scheme), Some(encoded), None) = (parts.next(), parts.next(), parts.next()) else {
+        return Credentials::Invalid;
+    };
+    if !scheme.eq_ignore_ascii_case("basic") {
+        return Credentials::Invalid;
+    }
+
+    let Ok(decoded) = BASE64_STANDARD.decode(encoded) else {
+        return Credentials::Invalid;
+    };
+    let decoded = Zeroizing::new(decoded);
+    let Some(separator) = decoded.iter().position(|byte| *byte == b':') else {
+        return Credentials::Invalid;
+    };
+    let Ok(username) = std::str::from_utf8(&decoded[..separator]) else {
+        return Credentials::Invalid;
+    };
+    let password = &decoded[separator + 1..];
+    if username.is_empty()
+        || username.len() > MAX_USERNAME_LEN
+        || password.is_empty()
+        || password.len() > MAX_PASSWORD_LEN
+    {
+        return Credentials::Invalid;
+    }
+
+    Credentials::Present(BasicCredentials {
+        username: username.to_owned(),
+        password: Zeroizing::new(password.to_vec()),
+    })
+}
+
+pub fn hash_password(password: &[u8]) -> Result<String, BoxError> {
+    if password.is_empty() {
+        return Err("password must not be empty".into());
+    }
+    if password.len() > MAX_PASSWORD_LEN {
+        return Err("password must not exceed 1024 bytes".into());
+    }
+
+    Ok(Argon2::default().hash_password(password)?.to_string())
+}
+
+#[derive(Debug, Error)]
+pub enum AclError {
+    #[error("ACL rule names must not be empty")]
+    EmptyRuleName,
+    #[error("ACL rule {0:?} must have a user or network selector")]
+    EmptyRule(String),
+    #[error("ACL rule name {0:?} is duplicated")]
+    DuplicateRule(String),
+    #[error("ACL rule {rule:?} references unknown user {username:?}")]
+    UnknownUser { rule: String, username: String },
+    #[error("ACL rule {rule:?} has invalid network {network:?}")]
+    InvalidNetwork { rule: String, network: String },
+    #[error("ACL user {0:?} has an invalid or unsupported Argon2id password hash")]
+    InvalidPasswordHash(String),
+    #[error("ACL user {0:?} has unsafe Argon2id parameters")]
+    UnsafePasswordParameters(String),
+    #[error("password verification task failed")]
+    AuthenticationTask,
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use http::HeaderValue;
+
+    use super::*;
+    use crate::config::{RuleConfig, UserConfig};
+
+    const PASSWORD: &[u8] = b"correct horse battery staple";
+
+    fn user_rule(name: &str, username: &str, action: Action) -> RuleConfig {
+        RuleConfig {
+            name: name.to_owned(),
+            action,
+            users: vec![username.to_owned()],
+            networks: vec![],
+        }
+    }
+
+    fn user_acl(default_action: Action) -> Acl {
+        let password_hash = hash_password(PASSWORD).unwrap();
+        Acl::from_config(AclConfig {
+            default_action,
+            users: BTreeMap::from([("alice".to_owned(), UserConfig { password_hash })]),
+            rules: vec![user_rule("alice", "alice", Action::Allow)],
+        })
+        .unwrap()
+    }
+
+    fn headers(username: &str, password: &str) -> HeaderMap {
+        let encoded = BASE64_STANDARD.encode(format!("{username}:{password}"));
+        HeaderMap::from_iter([(
+            PROXY_AUTHORIZATION,
+            HeaderValue::from_str(&format!("Basic {encoded}")).unwrap(),
+        )])
+    }
+
+    #[tokio::test]
+    async fn authenticates_basic_user() {
+        let acl = user_acl(Action::Deny);
+        let outcome = acl
+            .authorize(
+                "127.0.0.1".parse().unwrap(),
+                &headers("alice", "correct horse battery staple"),
+            )
+            .await
+            .unwrap();
+        assert_eq!(outcome.decision, Decision::Allow);
+        assert_eq!(outcome.reason, "user");
+    }
+
+    #[tokio::test]
+    async fn challenges_missing_malformed_unknown_and_wrong_credentials() {
+        let acl = user_acl(Action::Deny);
+        let malformed = HeaderMap::from_iter([(
+            PROXY_AUTHORIZATION,
+            HeaderValue::from_static("Basic not-valid-base64!"),
+        )]);
+        for headers in [
+            HeaderMap::new(),
+            headers("bob", "correct horse battery staple"),
+            headers("alice", "wrong"),
+            malformed,
+        ] {
+            let outcome = acl
+                .authorize("127.0.0.1".parse().unwrap(), &headers)
+                .await
+                .unwrap();
+            assert_eq!(outcome.decision, Decision::Challenge);
+        }
+    }
+
+    #[tokio::test]
+    async fn matches_ip_network_without_credentials() {
+        let acl = Acl::from_config(AclConfig {
+            default_action: Action::Deny,
+            users: BTreeMap::new(),
+            rules: vec![RuleConfig {
+                name: "localhost".to_owned(),
+                action: Action::Allow,
+                users: vec![],
+                networks: vec!["127.0.0.0/8".to_owned(), "::1/128".to_owned()],
+            }],
+        })
+        .unwrap();
+
+        let allowed = acl
+            .authorize("127.0.0.2".parse().unwrap(), &HeaderMap::new())
+            .await
+            .unwrap();
+        assert_eq!(allowed.decision, Decision::Allow);
+
+        let denied = acl
+            .authorize("192.0.2.1".parse().unwrap(), &HeaderMap::new())
+            .await
+            .unwrap();
+        assert_eq!(denied.decision, Decision::Deny);
+    }
+
+    #[tokio::test]
+    async fn applies_first_match_and_combines_selectors() {
+        let password_hash = hash_password(PASSWORD).unwrap();
+        let acl = Acl::from_config(AclConfig {
+            default_action: Action::Allow,
+            users: BTreeMap::from([("alice".to_owned(), UserConfig { password_hash })]),
+            rules: vec![
+                RuleConfig {
+                    name: "deny-alice-off-network".to_owned(),
+                    action: Action::Deny,
+                    users: vec!["alice".to_owned()],
+                    networks: vec!["192.0.2.0/24".to_owned()],
+                },
+                user_rule("allow-alice", "alice", Action::Allow),
+            ],
+        })
+        .unwrap();
+
+        let denied = acl
+            .authorize(
+                "192.0.2.10".parse().unwrap(),
+                &headers("alice", "correct horse battery staple"),
+            )
+            .await
+            .unwrap();
+        assert_eq!(denied.decision, Decision::Deny);
+        assert_eq!(denied.rule.as_deref(), Some("deny-alice-off-network"));
+
+        let allowed = acl
+            .authorize(
+                "198.51.100.10".parse().unwrap(),
+                &headers("alice", "correct horse battery staple"),
+            )
+            .await
+            .unwrap();
+        assert_eq!(allowed.decision, Decision::Allow);
+    }
+
+    #[test]
+    fn generates_salted_hashes_that_verify() {
+        let first = hash_password(PASSWORD).unwrap();
+        let second = hash_password(PASSWORD).unwrap();
+        assert_ne!(first, second);
+        assert!(first.starts_with("$argon2id$v=19$"));
+
+        for encoded in [first, second] {
+            let parsed = PasswordHash::new(&encoded).unwrap();
+            assert!(Argon2::default().verify_password(PASSWORD, &parsed).is_ok());
+            assert!(
+                Argon2::default()
+                    .verify_password(b"wrong", &parsed)
+                    .is_err()
+            );
+        }
+    }
+
+    #[test]
+    fn rejects_invalid_rules_and_hashes() {
+        let empty = Acl::from_config(AclConfig {
+            default_action: Action::Deny,
+            users: BTreeMap::new(),
+            rules: vec![RuleConfig {
+                name: "accidental-match-all".to_owned(),
+                action: Action::Allow,
+                users: vec![],
+                networks: vec![],
+            }],
+        });
+        assert!(matches!(empty, Err(AclError::EmptyRule(_))));
+
+        let invalid_hash = Acl::from_config(AclConfig {
+            default_action: Action::Deny,
+            users: BTreeMap::from([(
+                "alice".to_owned(),
+                UserConfig {
+                    password_hash: "plaintext".to_owned(),
+                },
+            )]),
+            rules: vec![],
+        });
+        assert!(matches!(
+            invalid_hash,
+            Err(AclError::InvalidPasswordHash(_))
+        ));
+    }
+}

src/config.rs

@@ -1,61 +1,153 @@
-use std::{env, net::SocketAddr, num::ParseIntError, time::Duration};
+use std::{collections::BTreeMap, fs, net::SocketAddr, path::Path, sync::Arc, time::Duration};
 
+use serde::Deserialize;
 use thiserror::Error;
 
-const DEFAULT_LISTEN_ADDR: &str = "127.0.0.1:8080";
+use crate::acl::{Acl, AclError};
+
 const DEFAULT_CONNECT_TIMEOUT_MS: u64 = 10_000;
 const DEFAULT_UPSTREAM_TIMEOUT_MS: u64 = 30_000;
 
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct Config {
     pub listen_addr: SocketAddr,
     pub connect_timeout: Duration,
     pub upstream_timeout: Duration,
+    pub(crate) acl: Arc<Acl>,
 }
 
 impl Config {
-    pub fn from_env() -> Result<Self, ConfigError> {
-        let listen_addr = env::var("HERMES_LISTEN_ADDR")
-            .unwrap_or_else(|_| DEFAULT_LISTEN_ADDR.to_owned())
-            .parse()?;
-        let connect_timeout =
-            duration_from_env("HERMES_CONNECT_TIMEOUT_MS", DEFAULT_CONNECT_TIMEOUT_MS)?;
-        let upstream_timeout =
-            duration_from_env("HERMES_UPSTREAM_TIMEOUT_MS", DEFAULT_UPSTREAM_TIMEOUT_MS)?;
+    pub fn load(path: impl AsRef<Path>) -> Result<Self, ConfigError> {
+        let path = path.as_ref();
+        let yaml = fs::read_to_string(path).map_err(|source| ConfigError::Read {
+            path: path.display().to_string(),
+            source,
+        })?;
+        Self::from_yaml(&yaml)
+    }
+
+    fn from_yaml(yaml: &str) -> Result<Self, ConfigError> {
+        let file: FileConfig = serde_norway::from_str(yaml)?;
+        if file.connect_timeout_ms == 0 {
+            return Err(ConfigError::ZeroDuration("connect_timeout_ms"));
+        }
+        if file.upstream_timeout_ms == 0 {
+            return Err(ConfigError::ZeroDuration("upstream_timeout_ms"));
+        }
 
         Ok(Self {
-            listen_addr,
-            connect_timeout,
-            upstream_timeout,
+            listen_addr: file.listen_addr,
+            connect_timeout: Duration::from_millis(file.connect_timeout_ms),
+            upstream_timeout: Duration::from_millis(file.upstream_timeout_ms),
+            acl: Arc::new(Acl::from_config(file.acl)?),
         })
     }
 }
 
-fn duration_from_env(name: &'static str, default_ms: u64) -> Result<Duration, ConfigError> {
-    let milliseconds = match env::var(name) {
-        Ok(value) => value
-            .parse()
-            .map_err(|source| ConfigError::InvalidDuration { name, source })?,
-        Err(_) => default_ms,
-    };
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct FileConfig {
+    #[serde(default = "default_listen_addr")]
+    listen_addr: SocketAddr,
+    #[serde(default = "default_connect_timeout_ms")]
+    connect_timeout_ms: u64,
+    #[serde(default = "default_upstream_timeout_ms")]
+    upstream_timeout_ms: u64,
+    acl: AclConfig,
+}
 
-    if milliseconds == 0 {
-        return Err(ConfigError::ZeroDuration(name));
-    }
+fn default_listen_addr() -> SocketAddr {
+    SocketAddr::from(([127, 0, 0, 1], 8080))
+}
 
-    Ok(Duration::from_millis(milliseconds))
+const fn default_connect_timeout_ms() -> u64 {
+    DEFAULT_CONNECT_TIMEOUT_MS
+}
+
+const fn default_upstream_timeout_ms() -> u64 {
+    DEFAULT_UPSTREAM_TIMEOUT_MS
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct AclConfig {
+    #[serde(default)]
+    pub default_action: Action,
+    #[serde(default)]
+    pub users: BTreeMap<String, UserConfig>,
+    #[serde(default)]
+    pub rules: Vec<RuleConfig>,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct UserConfig {
+    pub password_hash: String,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct RuleConfig {
+    pub name: String,
+    pub action: Action,
+    #[serde(default)]
+    pub users: Vec<String>,
+    #[serde(default)]
+    pub networks: Vec<String>,
+}
+
+#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)]
+#[serde(rename_all = "lowercase")]
+pub(crate) enum Action {
+    Allow,
+    #[default]
+    Deny,
 }
 
 #[derive(Debug, Error)]
 pub enum ConfigError {
-    #[error("invalid HERMES_LISTEN_ADDR: {0}")]
-    InvalidListenAddr(#[from] std::net::AddrParseError),
-    #[error("invalid {name}: {source}")]
-    InvalidDuration {
-        name: &'static str,
+    #[error("failed to read configuration {path}: {source}")]
+    Read {
+        path: String,
         #[source]
-        source: ParseIntError,
+        source: std::io::Error,
     },
+    #[error("invalid YAML configuration: {0}")]
+    Yaml(#[from] serde_norway::Error),
     #[error("{0} must be greater than zero")]
     ZeroDuration(&'static str),
+    #[error(transparent)]
+    Acl(#[from] AclError),
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn loads_defaults_and_localhost_acl() {
+        let config = Config::from_yaml(
+            r#"
+acl:
+  rules:
+    - name: localhost
+      action: allow
+      networks: ["127.0.0.1/32", "::1/128"]
+"#,
+        )
+        .unwrap();
+
+        assert_eq!(config.listen_addr, default_listen_addr());
+        assert_eq!(config.connect_timeout, Duration::from_secs(10));
+        assert_eq!(config.upstream_timeout, Duration::from_secs(30));
+    }
+
+    #[test]
+    fn rejects_unknown_fields_and_zero_timeouts() {
+        let unknown = Config::from_yaml("acl: {}\nunknown: true\n");
+        assert!(unknown.is_err());
+
+        let zero = Config::from_yaml("connect_timeout_ms: 0\nacl: {}\n");
+        assert!(matches!(zero, Err(ConfigError::ZeroDuration(_))));
+    }
 }

src/lib.rs

@@ -1,3 +1,4 @@
+pub mod acl;
 pub mod config;
 pub mod telemetry;
 

src/main.rs

@@ -1,11 +1,43 @@
-use hermes::{BoxError, config::Config, telemetry};
+use std::path::PathBuf;
+
+use clap::{Parser, Subcommand};
+use hermes::{BoxError, acl, config::Config, telemetry};
+use zeroize::Zeroizing;
+
+#[derive(Parser)]
+#[command(
+    version,
+    about = "A minimal HTTP forward proxy with OpenTelemetry observability"
+)]
+struct Cli {
+    #[arg(short, long, default_value = "hermes.yaml", value_name = "PATH")]
+    config: PathBuf,
+    #[command(subcommand)]
+    command: Option<Command>,
+}
+
+#[derive(Subcommand)]
+enum Command {
+    /// Generate a salted Argon2id password hash for the YAML configuration.
+    HashPassword,
+}
 
 #[tokio::main]
 async fn main() -> Result<(), BoxError> {
-    let telemetry = telemetry::init()?;
+    let cli = Cli::parse();
+    if matches!(cli.command, Some(Command::HashPassword)) {
+        let password = Zeroizing::new(rpassword::prompt_password("New password: ")?);
+        let confirmation = Zeroizing::new(rpassword::prompt_password("Confirm password: ")?);
+        if password.as_str() != confirmation.as_str() {
+            return Err("passwords do not match".into());
+        }
+        println!("{}", acl::hash_password(password.as_bytes())?);
+        return Ok(());
+    }
 
+    let telemetry = telemetry::init()?;
     let result: Result<(), BoxError> = async {
-        let config = Config::from_env()?;
+        let config = Config::load(&cli.config)?;
         hermes::run(config).await
     }
     .await;

src/metrics.rs

@@ -15,6 +15,7 @@ pub(crate) struct Metrics {
     active_tunnels: UpDownCounter<i64>,
     tunnel_duration: Histogram<f64>,
     tunnel_bytes: Counter<u64>,
+    acl_decisions: Counter<u64>,
 }
 
 impl Metrics {
@@ -60,6 +61,11 @@ impl Metrics {
                 .with_description("Bytes relayed through completed CONNECT tunnels")
                 .with_unit("By")
                 .build(),
+            acl_decisions: meter
+                .u64_counter("hermes.proxy.acl.decisions")
+                .with_description("Number of ACL decisions by outcome and reason")
+                .with_unit("{decision}")
+                .build(),
         }
     }
 
@@ -122,6 +128,16 @@ impl Metrics {
         );
     }
 
+    pub fn acl_decision(&self, decision: &str, reason: &str) {
+        self.acl_decisions.add(
+            1,
+            &[
+                KeyValue::new("acl.decision", decision.to_owned()),
+                KeyValue::new("acl.reason", reason.to_owned()),
+            ],
+        );
+    }
+
     pub fn tunnel_started(&self) -> ActiveTunnel {
         self.active_tunnels.add(1, &[]);
         ActiveTunnel {

src/proxy.rs

@@ -1,7 +1,16 @@
-use std::{convert::Infallible, net::SocketAddr, time::Instant};
+use std::{
+    convert::Infallible,
+    net::{IpAddr, SocketAddr},
+    sync::Arc,
+    time::Instant,
+};
 
 use bytes::Bytes;
-use http::{HeaderValue, Method, Request, Response, StatusCode, Uri, header::HOST, uri::Authority};
+use http::{
+    HeaderValue, Method, Request, Response, StatusCode, Uri,
+    header::{HOST, PROXY_AUTHENTICATE},
+    uri::Authority,
+};
 use http_body_util::{BodyExt, Empty, Full, combinators::UnsyncBoxBody};
 use hyper::{body::Incoming, server::conn::http1, service::service_fn};
 use hyper_util::{
@@ -14,11 +23,12 @@ use tokio::{
     task::JoinSet,
     time::timeout,
 };
-use tracing::{Instrument, Span, debug, error, field, info, info_span};
+use tracing::{Instrument, Span, debug, error, field, info, info_span, warn};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 
 use crate::{
     BoxError,
+    acl::{Acl, Decision},
     config::Config,
     metrics::Metrics,
     telemetry::{HeaderExtractor, HeaderInjector, remove_hop_by_hop_headers},
@@ -31,6 +41,7 @@ type HttpClient = Client<HttpConnector, Incoming>;
 #[derive(Clone)]
 struct ProxyState {
     client: HttpClient,
+    acl: Arc<Acl>,
     metrics: Metrics,
     connect_timeout: std::time::Duration,
     upstream_timeout: std::time::Duration,
@@ -44,6 +55,7 @@ impl ProxyState {
 
         Self {
             client: Client::builder(TokioExecutor::new()).build(connector),
+            acl: config.acl.clone(),
             metrics: Metrics::new(global::meter("hermes")),
             connect_timeout: config.connect_timeout,
             upstream_timeout: config.upstream_timeout,
@@ -92,9 +104,10 @@ pub async fn run(config: Config) -> Result<(), BoxError> {
 }
 
 async fn serve_connection(stream: TcpStream, peer: SocketAddr, state: ProxyState) {
+    let peer_ip = peer.ip();
     let service = service_fn(move |request| {
         let state = state.clone();
-        async move { Ok::<_, Infallible>(handle_request(request, state).await) }
+        async move { Ok::<_, Infallible>(handle_request(request, peer_ip, state).await) }
     });
 
     let connection = http1::Builder::new()
@@ -110,15 +123,19 @@ async fn serve_connection(stream: TcpStream, peer: SocketAddr, state: ProxyState
     }
 }
 
-async fn handle_request(request: Request<Incoming>, state: ProxyState) -> ProxyResponse {
+async fn handle_request(
+    request: Request<Incoming>,
+    peer: IpAddr,
+    state: ProxyState,
+) -> ProxyResponse {
     if request.method() == Method::CONNECT {
-        handle_connect(request, state).await
+        handle_connect(request, peer, state).await
     } else {
-        handle_http(request, state).await
+        handle_http(request, peer, state).await
     }
 }
 
-async fn handle_http(request: Request<Incoming>, state: ProxyState) -> ProxyResponse {
+async fn handle_http(request: Request<Incoming>, peer: IpAddr, state: ProxyState) -> ProxyResponse {
     let method = request.method().to_string();
     let uri = request.uri().clone();
     let parent = global::get_text_map_propagator(|propagator| {
@@ -134,6 +151,9 @@ async fn handle_http(request: Request<Incoming>, state: ProxyState) -> ProxyResp
         server.address = %uri.authority().map(Authority::host).unwrap_or_default(),
         http.response.status_code = field::Empty,
         error.type = field::Empty,
+        acl.decision = field::Empty,
+        acl.reason = field::Empty,
+        acl.rule = field::Empty,
         otel.status_code = field::Empty,
     );
     let _ = span.set_parent(parent);
@@ -141,13 +161,73 @@ async fn handle_http(request: Request<Incoming>, state: ProxyState) -> ProxyResp
     async move {
         let _active = state.metrics.request_started();
         let started = Instant::now();
-        let result = forward_http(request, &state).await;
+        let result = match authorize_request(&request, peer, &state).await {
+            Ok(Some(response)) => Ok(response),
+            Ok(None) => forward_http(request, &state).await,
+            Err(error) => Err(error),
+        };
         finish_request(result, &method, started, &state.metrics)
     }
     .instrument(span)
     .await
 }
 
+async fn authorize_request(
+    request: &Request<Incoming>,
+    peer: IpAddr,
+    state: &ProxyState,
+) -> Result<Option<ProxyResponse>, ProxyError> {
+    let outcome = state
+        .acl
+        .authorize(peer, request.headers())
+        .await
+        .map_err(|error| {
+            ProxyError::new(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "acl_internal",
+                "acl",
+                error.to_string(),
+            )
+        })?;
+
+    state
+        .metrics
+        .acl_decision(outcome.decision.as_str(), outcome.reason);
+    Span::current().record("acl.decision", outcome.decision.as_str());
+    Span::current().record("acl.reason", outcome.reason);
+    if let Some(rule) = &outcome.rule {
+        Span::current().record("acl.rule", rule);
+    }
+
+    match outcome.decision {
+        Decision::Allow => {
+            debug!(
+                acl.reason = outcome.reason,
+                acl.rule = outcome.rule.as_deref().unwrap_or_default(),
+                "ACL allowed request"
+            );
+            Ok(None)
+        }
+        Decision::Deny => {
+            warn!(
+                acl.reason = outcome.reason,
+                acl.rule = outcome.rule.as_deref().unwrap_or_default(),
+                "ACL denied request"
+            );
+            Ok(Some(error_response(StatusCode::FORBIDDEN)))
+        }
+        Decision::Challenge => {
+            info!(acl.reason = outcome.reason, "ACL challenged request");
+            let mut response = error_response(StatusCode::PROXY_AUTHENTICATION_REQUIRED);
+            response.headers_mut().insert(
+                PROXY_AUTHENTICATE,
+                HeaderValue::from_static("Basic realm=\"hermes\", charset=\"UTF-8\""),
+            );
+            Ok(Some(response))
+        }
+    }
+}
+
 async fn forward_http(
     mut request: Request<Incoming>,
     state: &ProxyState,
@@ -238,7 +318,11 @@ async fn send_upstream(
     .await
 }
 
-async fn handle_connect(request: Request<Incoming>, state: ProxyState) -> ProxyResponse {
+async fn handle_connect(
+    request: Request<Incoming>,
+    peer: IpAddr,
+    state: ProxyState,
+) -> ProxyResponse {
     let authority = request.uri().authority().cloned();
     let parent = global::get_text_map_propagator(|propagator| {
         propagator.extract(&HeaderExtractor(request.headers()))
@@ -254,6 +338,9 @@ async fn handle_connect(request: Request<Incoming>, state: ProxyState) -> ProxyR
         error.type = field::Empty,
         tunnel.bytes.client_to_upstream = field::Empty,
         tunnel.bytes.upstream_to_client = field::Empty,
+        acl.decision = field::Empty,
+        acl.reason = field::Empty,
+        acl.rule = field::Empty,
         otel.status_code = field::Empty,
     );
     let _ = span.set_parent(parent);
@@ -261,7 +348,11 @@ async fn handle_connect(request: Request<Incoming>, state: ProxyState) -> ProxyR
     async move {
         let _active = state.metrics.request_started();
         let started = Instant::now();
-        let result = prepare_connect(request, &state).await;
+        let result = match authorize_request(&request, peer, &state).await {
+            Ok(Some(response)) => Ok(response),
+            Ok(None) => prepare_connect(request, &state).await,
+            Err(error) => Err(error),
+        };
         finish_request(result, "CONNECT", started, &state.metrics)
     }
     .instrument(span)
@@ -514,10 +605,13 @@ impl ProxyError {
 #[cfg(test)]
 mod tests {
     use std::{
+        collections::BTreeMap,
         sync::{Arc, Mutex},
         time::Duration,
     };
 
+    use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
+    use http::header::PROXY_AUTHORIZATION;
     use hyper::client::conn::http1 as client_http1;
     use tokio::{
         io::{AsyncReadExt, AsyncWriteExt},
@@ -525,6 +619,7 @@ mod tests {
     };
 
     use super::*;
+    use crate::config::{AclConfig, Action, RuleConfig, UserConfig};
 
     #[test]
     fn validates_absolute_http_targets() {
@@ -583,13 +678,13 @@ mod tests {
                 .unwrap();
         });
 
-        let (proxy_addr, proxy_task) = spawn_proxy().await;
+        let (proxy_addr, proxy_task) = spawn_proxy_with_acl(basic_acl()).await;
         let (mut sender, connection_task) = connect_client(proxy_addr).await;
         let request = Request::builder()
             .method(Method::GET)
             .uri(format!("http://{upstream_addr}/hello?q=1"))
             .header(HOST, "wrong.invalid")
-            .header("proxy-authorization", "must-not-be-forwarded")
+            .header(PROXY_AUTHORIZATION, basic_authorization())
             .body(Full::new(Bytes::new()))
             .unwrap();
 
@@ -622,7 +717,7 @@ mod tests {
             stream.shutdown().await.unwrap();
         });
 
-        let (proxy_addr, proxy_task) = spawn_proxy().await;
+        let (proxy_addr, proxy_task) = spawn_proxy_with_acl(localhost_acl()).await;
         let (mut sender, connection_task) = connect_client(proxy_addr).await;
         let request = Request::builder()
             .method(Method::CONNECT)
@@ -645,18 +740,86 @@ mod tests {
         connection_task.abort();
     }
 
-    fn test_state() -> ProxyState {
-        ProxyState::new(&Config {
-            listen_addr: "127.0.0.1:0".parse().unwrap(),
-            connect_timeout: Duration::from_secs(2),
-            upstream_timeout: Duration::from_secs(2),
-        })
+    #[tokio::test]
+    async fn challenges_unauthenticated_http_and_connect_requests() {
+        let (proxy_addr, proxy_task) = spawn_proxy_with_acl(basic_acl()).await;
+        let (mut sender, connection_task) = connect_client(proxy_addr).await;
+
+        for request in [
+            Request::builder()
+                .method(Method::GET)
+                .uri("http://192.0.2.1/")
+                .body(Full::new(Bytes::new()))
+                .unwrap(),
+            Request::builder()
+                .method(Method::CONNECT)
+                .uri("192.0.2.1:443")
+                .body(Full::new(Bytes::new()))
+                .unwrap(),
+        ] {
+            let response = sender.send_request(request).await.unwrap();
+            assert_eq!(response.status(), StatusCode::PROXY_AUTHENTICATION_REQUIRED);
+            assert_eq!(
+                response.headers()[PROXY_AUTHENTICATE],
+                "Basic realm=\"hermes\", charset=\"UTF-8\""
+            );
+            response.into_body().collect().await.unwrap();
+        }
+
+        proxy_task.abort();
+        connection_task.abort();
     }
 
-    async fn spawn_proxy() -> (SocketAddr, JoinHandle<()>) {
+    fn basic_authorization() -> String {
+        format!(
+            "Basic {}",
+            BASE64_STANDARD.encode("alice:correct horse battery staple")
+        )
+    }
+
+    fn basic_acl() -> Acl {
+        Acl::from_config(AclConfig {
+            default_action: Action::Deny,
+            users: BTreeMap::from([(
+                "alice".to_owned(),
+                UserConfig {
+                    password_hash: crate::acl::hash_password(b"correct horse battery staple")
+                        .unwrap(),
+                },
+            )]),
+            rules: vec![RuleConfig {
+                name: "alice".to_owned(),
+                action: Action::Allow,
+                users: vec!["alice".to_owned()],
+                networks: vec![],
+            }],
+        })
+        .unwrap()
+    }
+
+    fn localhost_acl() -> Acl {
+        Acl::from_config(AclConfig {
+            default_action: Action::Deny,
+            users: BTreeMap::new(),
+            rules: vec![RuleConfig {
+                name: "localhost".to_owned(),
+                action: Action::Allow,
+                users: vec![],
+                networks: vec!["127.0.0.1/32".to_owned(), "::1/128".to_owned()],
+            }],
+        })
+        .unwrap()
+    }
+
+    async fn spawn_proxy_with_acl(acl: Acl) -> (SocketAddr, JoinHandle<()>) {
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
         let address = listener.local_addr().unwrap();
-        let state = test_state();
+        let state = ProxyState::new(&Config {
+            listen_addr: address,
+            connect_timeout: Duration::from_secs(2),
+            upstream_timeout: Duration::from_secs(2),
+            acl: Arc::new(acl),
+        });
         let task = tokio::spawn(async move {
             while let Ok((stream, peer)) = listener.accept().await {
                 let state = state.clone();