airs-cli 0.5.0

//! Endpoint admission and SDK client construction (DESIGN §5).
//!
//! Every configured URL is admitted with `Endpoint::loopback` (https anywhere, http
//! only for loopback hosts). Credentials come only from the tenant file and are
//! checked locally before any request, so configuration mistakes cost zero requests.
//!
//! A client with a plain-http (loopback) endpoint uses [`DirectTransport`], which never
//! goes through a proxy: http is admitted only because the traffic stays on the host,
//! and a proxy would carry the cleartext credentials off it.

use crate::debug::{DebugLog, DebugTransport};
use crate::error::{self, CliError, Operation, Phase, Plane, SdkContext, SecretSet};
use crate::tenant::config::{TenantConfig, js_trim};
use futures_util::StreamExt;
use futures_util::future::BoxFuture;
use scm::airs::management::ManagementClient;
use scm::airs::scan::Scanner;
use scm::auth::{AuthContext, Authentication, OAuthProvider, TokenLease, TokenProvider};
use scm::config::DestinationPolicy;
use scm::transport::{ReqwestTransport, Transport, TransportRequest, TransportResponse};
use scm::{Client, Endpoint, ErrorKind, Limits, OriginPolicy, RequestOptions, Secret, Session};
use std::sync::Arc;
use std::sync::atomic::{AtomicU16, Ordering};
use std::time::Duration;

pub const DEFAULT_TOKEN_ENDPOINT: &str =
    "https://auth.apps.paloaltonetworks.com/oauth2/access_token";
pub const DEFAULT_MGMT_ENDPOINT: &str = "https://api.apps.paloaltonetworks.com/aisec";
pub const DEFAULT_SCAN_ENDPOINT: &str = "https://service.api.aisecurity.paloaltonetworks.com";

/// The SDK default budget for one request without explicit `RequestOptions`.
pub const SINGLE_REQUEST_BUDGET: Duration = Duration::from_secs(60);
/// The SDK's fixed budget for an OAuth refresh when no options are passed.
pub const TOKEN_BUDGET: Duration = Duration::from_secs(30);
/// Full walks (`--all`, lookups, latest-topic listings).
pub const WALK_BUDGET: Duration = Duration::from_secs(300);
/// The whole of one doctor probe (token + request).
pub const PROBE_BUDGET: Duration = Duration::from_secs(5);

/// TS SDK default attempt count for 502/503/504 on management reads.
const MANAGEMENT_RETRIES: u32 = 5;
/// The TypeScript SDK's scanner credential limit (UTF-16 code units).
const MAX_SCANNER_CREDENTIAL: usize = 2048;

/// A client could not be configured; `key` names the tenant setting at fault.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SetupError {
    pub key: Option<&'static str>,
    pub error: CliError,
}

impl SetupError {
    fn config(key: &'static str, message: impl Into<String>) -> Self {
        Self {
            key: Some(key),
            error: CliError::operational(message),
        }
    }
    fn sdk(message: &str) -> Self {
        Self {
            key: None,
            error: CliError::operational(message),
        }
    }
    /// The failure message (for doctor details).
    pub fn message(&self) -> String {
        self.error.message()
    }
}

impl From<SetupError> for CliError {
    fn from(value: SetupError) -> Self {
        value.error
    }
}

/// The admission failure for `key` (the value is never echoed).
fn invalid_endpoint(key: &'static str) -> SetupError {
    SetupError::config(
        key,
        format!(
            "Invalid {key}: endpoints must use https (http is allowed only for loopback hosts) and must not contain credentials, a query or a fragment"
        ),
    )
}

/// Admit a configured URL (or the default when unset).
pub fn admit(
    key: &'static str,
    configured: Option<&str>,
    default: &str,
) -> Result<Endpoint, SetupError> {
    Endpoint::loopback(configured.unwrap_or(default)).map_err(|_| invalid_endpoint(key))
}

/// The status of the last HTTP response a client received (0 until one arrives).
///
/// A walk or probe that ends at its deadline while the SDK waits to retry a 429/5xx
/// surfaces as `Deadline`, and a deadline that drops a page fetch loses the answer's
/// status; doctor reads this to report what the endpoint actually said (§9).
#[derive(Clone, Debug, Default)]
pub struct LastStatus(Arc<AtomicU16>);

impl LastStatus {
    pub fn get(&self) -> Option<u16> {
        match self.0.load(Ordering::SeqCst) {
            0 => None,
            status => Some(status),
        }
    }
}

/// Records each response status in a [`LastStatus`]; otherwise the inner transport.
struct ObservedTransport {
    inner: Arc<dyn Transport>,
    last: LastStatus,
}

impl Transport for ObservedTransport {
    fn send<'a>(
        &'a self,
        request: TransportRequest,
    ) -> BoxFuture<'a, scm::Result<TransportResponse>> {
        Box::pin(async move {
            let response = self.inner.send(request).await?;
            self.last
                .0
                .store(response.status().as_u16(), Ordering::SeqCst);
            Ok(response)
        })
    }
}

/// The SDK's `ReqwestTransport` (which honours proxy settings), or [`DirectTransport`]
/// when any of the client's endpoints is plain http; wrapped by the debug transport
/// (a pass-through without `debug`, DESIGN §2.9) and observed by `last`.
fn transport(
    endpoints: &[&Endpoint],
    last: &LastStatus,
    debug: Option<&DebugLog>,
) -> Result<Arc<dyn Transport>, SetupError> {
    let failed = || SetupError::sdk("AISEC_SDK_ERROR:The HTTP transport could not be initialised");
    let inner: Arc<dyn Transport> = if endpoints
        .iter()
        .any(|endpoint| endpoint.url().scheme() == "http")
    {
        Arc::new(DirectTransport::new().ok_or_else(failed)?)
    } else {
        Arc::new(ReqwestTransport::new().map_err(|_| failed())?)
    };
    Ok(Arc::new(ObservedTransport {
        inner: DebugTransport::wrap(inner, debug),
        last: last.clone(),
    }))
}

/// The SDK's user agent (`scm-sdk` is pinned at `=0.1.0`).
const SDK_USER_AGENT: &str = "PAN-AIRS/0.1.0-rust-sdk";

/// `ReqwestTransport` with the SDK's hardening (no redirects, no retries, bounded
/// headers, gzip only) that never uses a proxy: environment and system proxies are
/// ignored. Used only for admitted plain-http (loopback) endpoints.
struct DirectTransport {
    client: reqwest::Client,
}

impl DirectTransport {
    fn new() -> Option<Self> {
        reqwest::Client::builder()
            .no_proxy()
            .redirect(reqwest::redirect::Policy::none())
            .retry(reqwest::retry::never())
            .connect_timeout(Duration::from_secs(10))
            .pool_max_idle_per_host(32)
            .pool_idle_timeout(Duration::from_secs(90))
            .http1_max_headers(128)
            .http2_max_header_list_size(16 * 1024)
            .no_brotli()
            .no_deflate()
            .no_zstd()
            .gzip(true)
            .referer(false)
            .user_agent(SDK_USER_AGENT)
            .build()
            .ok()
            .map(|client| Self { client })
    }
}

impl Transport for DirectTransport {
    fn send<'a>(
        &'a self,
        request: TransportRequest,
    ) -> BoxFuture<'a, scm::Result<TransportResponse>> {
        Box::pin(async move {
            let response = self
                .client
                .request(request.method().clone(), request.url().clone())
                .headers(request.headers().clone())
                .body(request.body().clone())
                .send()
                .await
                .map_err(|_| scm::Error::new(ErrorKind::Transport))?;
            let status = response.status();
            let headers = response.headers().clone();
            let stream = response
                .bytes_stream()
                .map(|chunk| chunk.map_err(|_| scm::Error::new(ErrorKind::Transport)));
            Ok(TransportResponse::new(status, headers, Box::pin(stream)))
        })
    }
}

/// The OAuth provider, recording every lease it hands out in the invocation's
/// [`SecretSet`]. The SDK fetches leases on its own (a 401/403 refresh-and-replay, an
/// expiry during a walk); each one is registered before any error is mapped, so a
/// server that echoes the bearer in a request ID or error code never gets it printed.
pub struct RecordingProvider {
    inner: Arc<OAuthProvider>,
    secrets: SecretSet,
    /// The invocation's API log: token records are held until the lease is registered.
    debug: Option<DebugLog>,
}

impl RecordingProvider {
    /// Fetch or reuse a lease under `options` (the SDK's 30 s refresh budget without).
    pub async fn get_token(&self, options: Option<&RequestOptions>) -> scm::Result<TokenLease> {
        let options = match options {
            Some(options) => options.clone(),
            None => RequestOptions::with_timeout(TOKEN_BUDGET)?,
        };
        self.token(&options).await
    }
}

impl TokenProvider for RecordingProvider {
    fn token<'a>(&'a self, options: &'a RequestOptions) -> BoxFuture<'a, scm::Result<TokenLease>> {
        Box::pin(async move {
            // The token response's record waits until its lease is in the set, so a
            // token echoed in that response's own headers is masked (DESIGN §2.9).
            // Dropping this future drops the hold, which withholds the headers.
            let hold = self.debug.as_ref().map(DebugLog::hold_tokens);
            let result = self.inner.token(options).await;
            if let Ok(lease) = &result {
                self.secrets.add(lease.secret().expose());
            }
            if let Some(hold) = hold {
                hold.release();
            }
            result
        })
    }
    fn context(&self) -> &AuthContext {
        self.inner.context()
    }
    fn invalidate(&self, generation: u64) -> bool {
        self.inner.invalidate(generation)
    }
}

/// The management client and the application-owned OAuth provider behind it.
pub struct ManagementClients {
    pub provider: Arc<RecordingProvider>,
    pub management: ManagementClient,
    /// The last response status either of them received.
    pub last_status: LastStatus,
}

impl ManagementClients {
    /// Build from the effective config. Credentials must already be checked present.
    /// Every OAuth lease is registered in `secrets` (shared with the invocation); every
    /// attempt of the client and its provider is recorded in `debug`.
    pub fn build(
        config: &TenantConfig,
        session: &Session,
        secrets: &SecretSet,
        debug: Option<&DebugLog>,
    ) -> Result<Self, SetupError> {
        let api = admit(
            "mgmtEndpoint",
            config.effective_str("mgmtEndpoint"),
            DEFAULT_MGMT_ENDPOINT,
        )?;
        let issuer = admit(
            "mgmtTokenEndpoint",
            config.effective_str("mgmtTokenEndpoint"),
            DEFAULT_TOKEN_ENDPOINT,
        )?;
        let tsg = config.effective_str("mgmtTsgId").unwrap_or_default();
        if tsg.is_empty() || tsg.len() > 32 || !tsg.bytes().all(|b| b.is_ascii_digit()) {
            return Err(SetupError::config(
                "mgmtTsgId",
                "Invalid mgmtTsgId: the Rust SDK requires 1-32 ASCII digits",
            ));
        }
        let client_id = config.effective_str("mgmtClientId").unwrap_or_default();
        let context = AuthContext::new(issuer.clone(), client_id, 0, tsg).map_err(|_| {
            SetupError::config(
                "mgmtClientId",
                "Invalid mgmtClientId: the Rust SDK requires at most 1024 bytes without control characters",
            )
        })?;
        let invalid_secret = || {
            SetupError::config(
                "mgmtClientSecret",
                "Invalid mgmtClientSecret: the Rust SDK requires at most 16 KiB without control characters",
            )
        };
        let secret = Secret::new(config.effective_str("mgmtClientSecret").unwrap_or_default())
            .map_err(|_| invalid_secret())?;
        let last_status = LastStatus::default();
        let transport = transport(&[&api, &issuer], &last_status, debug)?;
        let policy: Arc<dyn DestinationPolicy> = Arc::new(OriginPolicy::new([api.clone(), issuer]));
        let provider = Arc::new(RecordingProvider {
            inner: Arc::new(
                OAuthProvider::new(
                    context,
                    secret,
                    transport.clone(),
                    policy.clone(),
                    session.clone(),
                )
                .map_err(|_| invalid_secret())?,
            ),
            secrets: secrets.clone(),
            debug: debug.cloned(),
        });
        let mut limits = Limits::default();
        limits.max_retries = MANAGEMENT_RETRIES;
        let client = Client::builder(api)
            .transport(transport)
            .destination_policy(policy)
            .authentication(Authentication::oauth(
                provider.clone() as Arc<dyn TokenProvider>,
                false,
            ))
            .session(session.clone())
            .limits(limits)
            .build()
            .map_err(|_| {
                SetupError::sdk("AISEC_SDK_ERROR:The management client could not be configured")
            })?;
        let management = ManagementClient::new(client).map_err(|_| {
            SetupError::sdk("AISEC_SDK_ERROR:The management client could not be configured")
        })?;
        Ok(Self {
            provider,
            management,
            last_status,
        })
    }

    /// Fetch the OAuth token explicitly so token failures are reported as such.
    /// `budget` is the duration `options` was created with (30 s without options).
    pub async fn authenticate(
        &self,
        options: Option<&RequestOptions>,
        budget: Duration,
    ) -> Result<(), CliError> {
        match self.provider.get_token(options).await {
            Ok(_) => Ok(()),
            Err(failure) => Err(error::from_sdk(
                &failure,
                &SdkContext {
                    plane: Plane::Management,
                    phase: Phase::Token,
                    operation: Operation::Read,
                    timeout: budget,
                    secrets: &self.provider.secrets,
                },
            )),
        }
    }
}

/// A trimmed scanner credential, checked against the TypeScript and SDK limits.
fn scanner_credential(
    config: &TenantConfig,
    key: &'static str,
    label: &str,
) -> Result<Option<Secret>, SetupError> {
    let Some(value) = config
        .effective_str(key)
        .map(js_trim)
        .filter(|v| !v.is_empty())
    else {
        return Ok(None);
    };
    if value.encode_utf16().nth(MAX_SCANNER_CREDENTIAL).is_some() {
        return Err(SetupError::config(
            key,
            format!(
                "AISEC_USER_REQUEST_PAYLOAD_ERROR:{label} exceeds max length of {MAX_SCANNER_CREDENTIAL}"
            ),
        ));
    }
    if !value.bytes().all(|byte| byte.is_ascii_graphic()) {
        return Err(SetupError::config(
            key,
            format!("Invalid {key}: the Rust SDK requires printable ASCII without spaces"),
        ));
    }
    Secret::new(value)
        .map(Some)
        .map_err(|_| SetupError::config(key, format!("Invalid {key}")))
}

/// Build the scanner through `scm::configuration::init` (TS `init()` semantics); every
/// attempt is recorded in `debug`.
pub fn build_scanner(
    config: &TenantConfig,
    session: &Session,
    debug: Option<&DebugLog>,
) -> Result<Scanner, SetupError> {
    build_observed_scanner(config, session, debug).map(|(scanner, _)| scanner)
}

/// [`build_scanner`], plus the status of the last response it receives.
pub fn build_observed_scanner(
    config: &TenantConfig,
    session: &Session,
    debug: Option<&DebugLog>,
) -> Result<(Scanner, LastStatus), SetupError> {
    let endpoint = admit(
        "airsApiEndpoint",
        config.effective_str("airsApiEndpoint"),
        DEFAULT_SCAN_ENDPOINT,
    )?;
    let api_key = scanner_credential(config, "airsApiKey", "apiKey")?;
    let api_token = scanner_credential(config, "airsApiToken", "apiToken")?;
    let mut options = scm::configuration::InitOptions::default();
    options.api_key = api_key;
    options.api_token = api_token;
    options.num_retries = config.num_retries();
    let last_status = LastStatus::default();
    let transport = transport(&[&endpoint], &last_status, debug)?;
    options.api_endpoint = Some(endpoint);
    let configuration =
        scm::configuration::init(options).map_err(|_| invalid_endpoint("airsApiEndpoint"))?;
    let failed = || SetupError::sdk("AISEC_SDK_ERROR:The scanner client could not be configured");
    let client = configuration
        .client_builder()
        .map_err(|_| failed())?
        .transport(transport)
        .session(session.clone())
        .build()
        .map_err(|_| failed())?;
    Ok((Scanner::new(client).map_err(|_| failed())?, last_status))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::tenant::config::{ConfigValue, schema_key};

    /// A tenant config from a JSON object of strings and integers (test shorthand).
    fn config(text: &str) -> TenantConfig {
        let serde_json::Value::Object(entries) = serde_json::from_str(text).unwrap() else {
            panic!("object expected");
        };
        TenantConfig::from_values(entries.into_iter().map(|(key, value)| {
            let key = schema_key(&key).unwrap();
            let value = match value {
                serde_json::Value::String(text) => ConfigValue::Text(text),
                serde_json::Value::Number(number) => ConfigValue::Integer(number.as_i64().unwrap()),
                other => panic!("unsupported {other}"),
            };
            (key, value)
        }))
    }

    #[test]
    fn endpoint_admission() {
        for url in [
            "https://example.com/aisec",
            "http://127.0.0.1:8080/aisec",
            "http://127.9.9.9",
            "http://localhost:1",
            "http://[::1]:9/x",
        ] {
            assert!(
                admit("mgmtEndpoint", Some(url), DEFAULT_MGMT_ENDPOINT).is_ok(),
                "{url}"
            );
        }
        for url in [
            "http://example.test",
            "http://0.0.0.0:1",
            "https://user:pw@example.com",
            "https://example.com/?q=1",
            "https://example.com/#f",
            "not a url",
        ] {
            let error = admit("mgmtEndpoint", Some(url), DEFAULT_MGMT_ENDPOINT).unwrap_err();
            assert_eq!(error.key, Some("mgmtEndpoint"));
            assert!(!error.message().contains(url), "the value is never echoed");
            assert!(
                error
                    .message()
                    .starts_with("Invalid mgmtEndpoint: endpoints must use https")
            );
        }
        assert!(admit("x", None, DEFAULT_SCAN_ENDPOINT).is_ok());
    }

    #[test]
    fn management_configuration_errors_are_local() {
        let session = Session::new();
        let secrets = SecretSet::default();
        let bad_tsg = config(r#"{"mgmtClientId":"c","mgmtClientSecret":"s","mgmtTsgId":"abc"}"#);
        let error = ManagementClients::build(&bad_tsg, &session, &secrets, None)
            .err()
            .unwrap();
        assert_eq!(
            error.message(),
            "Invalid mgmtTsgId: the Rust SDK requires 1-32 ASCII digits"
        );
        let bad_endpoint = config(
            r#"{"mgmtClientId":"c","mgmtClientSecret":"s","mgmtTsgId":"1","mgmtTokenEndpoint":"http://example.test/t"}"#,
        );
        let error = ManagementClients::build(&bad_endpoint, &session, &secrets, None)
            .err()
            .unwrap();
        assert_eq!(error.key, Some("mgmtTokenEndpoint"));
        let control = config(r#"{"mgmtClientId":"c\n","mgmtClientSecret":"s","mgmtTsgId":"1"}"#);
        assert_eq!(
            ManagementClients::build(&control, &session, &secrets, None)
                .err()
                .unwrap()
                .key,
            Some("mgmtClientId")
        );
        let good = config(
            r#"{"mgmtClientId":"c","mgmtClientSecret":"s","mgmtTsgId":"1","mgmtEndpoint":"http://127.0.0.1:1/aisec","mgmtTokenEndpoint":"http://127.0.0.1:1/token"}"#,
        );
        assert!(ManagementClients::build(&good, &session, &secrets, None).is_ok());
    }

    #[test]
    fn scanner_credentials_are_trimmed_and_bounded() {
        let session = Session::new();
        let good = config(r#"{"airsApiKey":"  key-1 \n","airsApiEndpoint":"http://127.0.0.1:1"}"#);
        assert!(build_scanner(&good, &session, None).is_ok());
        let long = format!(r#"{{"airsApiKey":"{}"}}"#, "k".repeat(2049));
        assert_eq!(
            build_scanner(&config(&long), &session, None)
                .err()
                .unwrap()
                .message(),
            "AISEC_USER_REQUEST_PAYLOAD_ERROR:apiKey exceeds max length of 2048"
        );
        let exact = format!(r#"{{"airsApiToken":"{}"}}"#, "k".repeat(2048));
        assert!(build_scanner(&config(&exact), &session, None).is_ok());
        let spaced = config(r#"{"airsApiToken":"a b"}"#);
        assert_eq!(
            build_scanner(&spaced, &session, None).err().unwrap().key,
            Some("airsApiToken")
        );
        let endpoint = config(r#"{"airsApiKey":"k","airsApiEndpoint":"http://example.test"}"#);
        assert_eq!(
            build_scanner(&endpoint, &session, None).err().unwrap().key,
            Some("airsApiEndpoint")
        );
    }
}