fluidattacks_core_git/

aws.rs

use std::time::Duration;

use aws_config::timeout::TimeoutConfig;
use aws_sdk_sts::error::{ConnectorError, DisplayErrorContext, ProvideErrorMetadata, SdkError};
use aws_sdk_sts::operation::assume_role::{AssumeRoleError, AssumeRoleOutput};
use regex::Regex;
use uuid::Uuid;

use crate::errors::StsError;

/// STS answers in tenths of a second; this leaves room for a slow one, as git's
/// ls-remote does. Without one the SDK waits as long as STS holds the connection open.
const READ_TIMEOUT: Duration = Duration::from_secs(20);

pub struct AwsTempCreds {
    pub access_key: String,
    pub secret_key: String,
    pub session_token: String,
    pub region: String,
}

struct Hop<'a> {
    role_arn: &'a str,
    external_id: Option<&'a str>,
    policy: Option<&'static str>,
}

/// The intermediate session may only assume a role, so a chained caller never
/// holds the intermediate role's other permissions. The resource stays open
/// because a policy would match the client's ARN as typed, case included.
const ONLY_ASSUME_ROLE: &str = r#"{
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Action": "sts:AssumeRole",
        "Resource": "*"
    }]
}"#;

/// Chained when the caller is not the principal the client's trust policy pins.
const fn chain<'a>(
    arn: &'a str,
    external_id: &'a str,
    via: Option<&'a str>,
) -> (Hop<'a>, Option<Hop<'a>>) {
    let client = Hop {
        role_arn: arn,
        external_id: Some(external_id),
        policy: None,
    };

    match via {
        Some(intermediate) => (
            Hop {
                role_arn: intermediate,
                external_id: None,
                policy: Some(ONLY_ASSUME_ROLE),
            },
            Some(client),
        ),
        None => (client, None),
    }
}

pub async fn assume_role(
    arn: &str,
    external_id: &str,
    url: &str,
    via: Option<&str>,
) -> Result<AwsTempCreds, StsError> {
    let region = extract_region(url);
    let (first, then) = chain(arn, external_id, via);

    let mut creds = Sts::signed_as(&region, None).await.assume(&first).await?;
    if let Some(hop) = then {
        let chained = Sts::signed_as(&region, Some(&creds)).await;
        creds = chained.assume(&hop).await?;
    }

    Ok(creds)
}

/// STS in the repository's region, signed by the worker or by the previous hop's session.
struct Sts {
    client: aws_sdk_sts::Client,
    region: String,
}

impl Sts {
    async fn signed_as(region: &str, from: Option<&AwsTempCreds>) -> Self {
        Self {
            client: aws_sdk_sts::Client::new(&config(region, from).load().await),
            region: region.to_owned(),
        }
    }

    async fn assume(&self, hop: &Hop<'_>) -> Result<AwsTempCreds, StsError> {
        let mut request = self
            .client
            .assume_role()
            .role_arn(hop.role_arn)
            .role_session_name(format!("session-{}", Uuid::new_v4()));

        if let Some(external_id) = hop.external_id {
            request = request.external_id(external_id);
        }

        if let Some(policy) = hop.policy {
            request = request.policy(policy);
        }

        let resp = request
            .send()
            .await
            .map_err(|error| refusal(hop.role_arn, &error))?;

        session(hop.role_arn, &resp, &self.region)
    }
}

fn config(region: &str, from: Option<&AwsTempCreds>) -> aws_config::ConfigLoader {
    // Only the read timeout: the SDK's connect timeout stays.
    let timeouts = TimeoutConfig::builder().read_timeout(READ_TIMEOUT).build();
    let mut config = aws_config::defaults(aws_config::BehaviorVersion::latest())
        .region(aws_config::Region::new(region.to_owned()))
        .timeout_config(timeouts);

    if let Some(session) = from {
        config = config.credentials_provider(aws_sdk_sts::config::Credentials::new(
            session.access_key.clone(),
            session.secret_key.clone(),
            Some(session.session_token.clone()),
            None,
            "fluidattacks-core-chain",
        ));
    }

    config
}

// botocore says "Unknown" for a code or message STS left out.
fn refusal<R: std::fmt::Debug>(arn: &str, error: &SdkError<AssumeRoleError, R>) -> StsError {
    let arn = arn.to_owned();
    if let SdkError::ServiceError(refused) = error {
        return StsError::Refused {
            arn,
            code: refused.err().code().unwrap_or("Unknown").to_owned(),
            message: refused.err().message().unwrap_or("Unknown").to_owned(),
        };
    }

    let reason = DisplayErrorContext(error).to_string();
    let SdkError::DispatchFailure(failure) = error else {
        return StsError::Unreachable { arn, reason };
    };
    let Some(connector) = failure.as_connector_error() else {
        return StsError::Unreachable { arn, reason };
    };

    // Told apart as botocore tells its classes apart, since integrates catches those.
    let timed_out = connector.is_timeout();
    if never_connected(connector) {
        return if timed_out {
            StsError::ConnectTimeout { arn, reason }
        } else {
            StsError::NoConnection { arn, reason }
        };
    }
    if timed_out {
        StsError::ReadTimeout { arn, reason }
    } else if hung_up(connector) {
        StsError::ConnectionClosed { arn, reason }
    } else {
        StsError::Unreachable { arn, reason }
    }
}

fn never_connected(connector: &ConnectorError) -> bool {
    causes(connector).any(|cause| {
        cause
            .downcast_ref::<hyper_util::client::legacy::Error>()
            .is_some_and(hyper_util::client::legacy::Error::is_connect)
    })
}

fn hung_up(connector: &ConnectorError) -> bool {
    connector.is_io()
        || causes(connector).any(|cause| {
            cause
                .downcast_ref::<hyper::Error>()
                .is_some_and(|hyper| hyper.is_incomplete_message() || hyper.is_closed())
        })
}

fn causes<'a>(
    error: &'a (dyn std::error::Error + 'static),
) -> impl Iterator<Item = &'a (dyn std::error::Error + 'static)> {
    std::iter::successors(Some(error), |current| current.source())
}

fn session(arn: &str, resp: &AssumeRoleOutput, region: &str) -> Result<AwsTempCreds, StsError> {
    let no_credentials = || StsError::NoCredentials {
        arn: arn.to_owned(),
    };
    let creds = resp.credentials().ok_or_else(no_credentials)?;

    let access_key = creds.access_key_id().to_owned();
    let secret_key = creds.secret_access_key().to_owned();
    let session_token = creds.session_token().to_owned();

    if access_key.is_empty() || secret_key.is_empty() {
        return Err(no_credentials());
    }

    Ok(AwsTempCreds {
        access_key,
        secret_key,
        session_token,
        region: region.to_owned(),
    })
}

fn extract_region(url: &str) -> String {
    [
        r"codecommit::([a-z0-9-]+)://",
        r"git-codecommit\.([a-z0-9-]+)\.amazonaws\.com",
    ]
    .into_iter()
    .find_map(|pattern| {
        let region = Regex::new(pattern).ok()?.captures(url)?.get(1)?;
        Some(region.as_str().to_owned())
    })
    .unwrap_or_else(|| "us-east-1".to_owned())
}

#[cfg(test)]
mod tests {
    use aws_sdk_sts::error::{ErrorMetadata, SdkError};
    use aws_sdk_sts::operation::assume_role::{AssumeRoleError, AssumeRoleOutput};
    use aws_sdk_sts::primitives::DateTime;
    use aws_sdk_sts::types::Credentials;
    use aws_sdk_sts::Client;
    use std::io::{BufRead, BufReader, Read, Write};
    use std::net::{TcpListener, TcpStream};
    use std::time::Duration;

    use super::{chain, config, extract_region, refusal, session, Hop, Sts, TimeoutConfig};
    use crate::errors::StsError;

    const ROLE: &str = "arn:aws:iam::1:role/client";

    #[test]
    fn a_refusal_carries_the_code_and_message_sts_answered() {
        let refused = AssumeRoleError::generic(
            ErrorMetadata::builder()
                .code("AccessDenied")
                .message("not authorized to perform sts:AssumeRole")
                .build(),
        );

        let error = refusal(ROLE, &SdkError::service_error(refused, ()));

        assert_eq!(
            error.to_string(),
            "STS refused the role arn:aws:iam::1:role/client: AccessDenied: not authorized to perform sts:AssumeRole"
        );
    }

    #[test]
    fn an_sts_that_never_answered_is_not_a_refusal() {
        let error = refusal(
            ROLE,
            &SdkError::<AssumeRoleError, ()>::timeout_error("timed out"),
        );

        assert!(matches!(error, StsError::Unreachable { .. }), "{error:?}");
    }

    #[test]
    fn an_answer_without_a_usable_key_is_no_session() {
        let keys = |access: &str, secret: &str| {
            Credentials::builder()
                .access_key_id(access)
                .secret_access_key(secret)
                .session_token("a-token")
                .expiration(DateTime::from_secs(0))
                .build()
                .unwrap()
        };

        for answer in [
            AssumeRoleOutput::builder().build(),
            AssumeRoleOutput::builder()
                .credentials(keys("", "a-secret-key"))
                .build(),
            AssumeRoleOutput::builder()
                .credentials(keys("an-access-key", ""))
                .build(),
        ] {
            let outcome = session(ROLE, &answer, "us-east-1");

            assert!(
                matches!(outcome, Err(StsError::NoCredentials { .. })),
                "{answer:?}"
            );
        }
    }

    #[test]
    fn the_region_comes_from_the_url_or_falls_back_to_us_east_1() {
        for (url, region) in [
            ("codecommit::eu-west-1://a-repo", "eu-west-1"),
            (
                "https://git-codecommit.sa-east-1.amazonaws.com/v1/repos/a-repo",
                "sa-east-1",
            ),
            ("codecommit://a-repo", "us-east-1"),
        ] {
            assert_eq!(extract_region(url), region);
        }
    }

    #[test]
    fn one_call_takes_the_client_role_with_its_external_id() {
        let (first, then) = chain("arn:aws:iam::1:role/client", "an-external-id", None);

        assert_eq!(first.role_arn, "arn:aws:iam::1:role/client");
        assert_eq!(first.external_id, Some("an-external-id"));
        assert_eq!(first.policy, None);
        assert!(then.is_none());
    }

    #[test]
    fn a_chain_takes_the_intermediate_role_first_and_without_the_external_id() {
        let (first, then) = chain(
            "arn:aws:iam::1:role/client",
            "an-external-id",
            Some("arn:aws:iam::2:role/intermediate"),
        );

        assert_eq!(first.role_arn, "arn:aws:iam::2:role/intermediate");
        assert_eq!(first.external_id, None);

        let client = then.expect("the client role is still taken");
        assert_eq!(client.role_arn, "arn:aws:iam::1:role/client");
        assert_eq!(client.external_id, Some("an-external-id"));
        assert_eq!(client.policy, None);
    }

    #[test]
    fn a_chain_lets_the_intermediate_session_only_assume_a_role() {
        let (first, _) = chain(
            "arn:aws:iam::1:role/client",
            "an-external-id",
            Some("arn:aws:iam::2:role/intermediate"),
        );

        let policy: serde_json::Value =
            serde_json::from_str(first.policy.expect("the intermediate session is narrowed"))
                .expect("the policy is json");
        assert_eq!(
            policy.get("Statement"),
            Some(&serde_json::json!([{
                "Effect": "Allow",
                "Action": "sts:AssumeRole",
                "Resource": "*",
            }]))
        );
    }

    // An STS on a local port that does `answer` with each request it reads.
    fn sts(answer: fn(TcpStream)) -> String {
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
        let port = listener.local_addr().expect("addr").port();
        std::thread::spawn(move || {
            for stream in listener.incoming().flatten() {
                std::thread::spawn(move || serve(stream, answer));
            }
        });
        format!("http://127.0.0.1:{port}")
    }

    fn serve(stream: TcpStream, answer: fn(TcpStream)) {
        read_request(&stream);
        answer(stream);
    }

    fn read_request(stream: &TcpStream) {
        let mut reader = BufReader::new(stream);
        let mut length = 0;
        let mut line = String::new();
        while reader.read_line(&mut line).is_ok_and(|read| read > 2) {
            if let Some(value) = line.to_lowercase().strip_prefix("content-length: ") {
                length = value.trim().parse().unwrap_or(0);
            }
            line.clear();
        }
        let _ = reader.read_exact(&mut vec![0; length]);
    }

    async fn ask_sts(
        endpoint: &str,
        hop: &Hop<'_>,
        signed: bool,
    ) -> Result<super::AwsTempCreds, StsError> {
        let quick = TimeoutConfig::builder()
            .read_timeout(Duration::from_millis(200))
            .build();
        let mut loader = config("us-east-1", None)
            .endpoint_url(endpoint)
            .timeout_config(quick);
        loader = if signed {
            loader.credentials_provider(aws_sdk_sts::config::Credentials::new(
                "a-key", "a-secret", None, None, "test",
            ))
        } else {
            loader.no_credentials()
        };
        let sts = Sts {
            client: Client::new(&loader.load().await),
            region: "us-east-1".to_owned(),
        };
        sts.assume(hop).await
    }

    fn client_role() -> Hop<'static> {
        chain(ROLE, "an-external-id", None).0
    }

    #[tokio::test]
    async fn sts_has_twenty_seconds_to_answer() {
        let sts = Sts::signed_as("us-east-1", None).await;

        let read_timeout = sts
            .client
            .config()
            .timeout_config()
            .and_then(TimeoutConfig::read_timeout);

        assert_eq!(read_timeout, Some(Duration::from_secs(20)));
    }

    #[tokio::test]
    async fn an_sts_that_keeps_the_request_unanswered_is_a_read_timeout() {
        let endpoint = sts(|_stream| std::thread::sleep(Duration::from_secs(5)));

        let outcome = ask_sts(&endpoint, &client_role(), true).await;

        assert!(
            matches!(outcome, Err(StsError::ReadTimeout { .. })),
            "{:?}",
            outcome.err()
        );
    }

    #[tokio::test]
    async fn an_sts_that_closes_the_connection_unanswered_is_connection_closed() {
        let endpoint = sts(drop);

        let outcome = ask_sts(&endpoint, &client_role(), true).await;

        assert!(
            matches!(outcome, Err(StsError::ConnectionClosed { .. })),
            "{:?}",
            outcome.err()
        );
    }

    #[tokio::test]
    async fn an_sts_that_resets_the_connection_unanswered_is_connection_closed() {
        let endpoint = sts(|stream| {
            // With no linger, closing sends a reset instead of an orderly close.
            rustix::net::sockopt::set_socket_linger(&stream, Some(Duration::ZERO)).expect("linger");
        });

        let outcome = ask_sts(&endpoint, &client_role(), true).await;

        assert!(
            matches!(outcome, Err(StsError::ConnectionClosed { .. })),
            "{:?}",
            outcome.err()
        );
    }

    #[tokio::test]
    async fn a_port_nobody_listens_on_is_no_connection() {
        let closed = TcpListener::bind("127.0.0.1:0").expect("bind");
        let endpoint = format!("http://{}", closed.local_addr().expect("addr"));
        drop(closed);

        let outcome = ask_sts(&endpoint, &client_role(), true).await;

        assert!(
            matches!(outcome, Err(StsError::NoConnection { .. })),
            "{:?}",
            outcome.err()
        );
    }

    #[tokio::test]
    async fn a_caller_without_credentials_to_sign_with_is_unreachable() {
        let endpoint = sts(drop);

        let outcome = ask_sts(&endpoint, &client_role(), false).await;

        assert!(
            matches!(outcome, Err(StsError::Unreachable { .. })),
            "{:?}",
            outcome.err()
        );
    }

    #[tokio::test]
    async fn a_role_below_the_sts_limits_is_left_for_sts_to_refuse() {
        // botocore refused these before asking; the SDK sends them as they are.
        let endpoint = sts(|mut stream| {
            let body = "<ErrorResponse><Error><Code>ValidationError</Code>\
                        <Message>too short</Message></Error></ErrorResponse>";
            let _ = write!(
                stream,
                "HTTP/1.1 400 Bad Request\r\nContent-Type: text/xml\r\n\
                 Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
                body.len()
            );
        });
        let short = Hop {
            role_arn: "x",
            external_id: Some("a"),
            policy: None,
        };

        let outcome = ask_sts(&endpoint, &short, true).await;

        let Err(StsError::Refused { code, .. }) = outcome else {
            panic!("{:?}", outcome.err());
        };
        assert_eq!(code, "ValidationError");
    }
}