keyhog-sources 0.5.43

//! Slack source: fetches messages from Slack channels using the Web API.
//!
//! This allows KeyHog to identify secrets leaked in chat history.
//! Requires a Slack API token (Bot or User) with `channels:history` and `groups:history` scopes.

use keyhog_core::{Chunk, ChunkMetadata, Source, SourceError};
use reqwest::blocking::{Client, Response};
use serde::{de::DeserializeOwned, Deserialize};

/// Scan Slack messages via the `conversations.history` API.
pub struct SlackSource {
    token: String,
    lookback_messages: usize,
    endpoint: String,
    limits: crate::SourceLimits,
    /// Shared HTTP policy (proxy, insecure_tls, ua_suffix, timeout). Defaults to
    /// the explicit-only `HttpClientConfig` policy; no environment variable can
    /// reroute Slack API calls. Set via `with_http_config` so the CLI's
    /// `--proxy` / `--insecure` reach this source instead of bypassing the
    /// configured corporate proxy or the operator's Burp interception.
    http: crate::http::HttpClientConfig,
}

impl SlackSource {
    /// Create a new Slack source.
    pub fn new(token: impl Into<String>) -> Self {
        Self {
            token: token.into(),
            lookback_messages: 1000,
            endpoint: "https://slack.com/api".into(),
            limits: crate::SourceLimits::default(),
            http: crate::http::HttpClientConfig {
                ua_suffix: Some("slack".into()),
                ..Default::default()
            },
        }
    }

    /// Override the shared HTTP policy. Threads CLI `--proxy` / `--insecure`
    /// into the Slack API client.
    pub(crate) fn with_http_config(mut self, http: crate::http::HttpClientConfig) -> Self {
        self.http = http;
        self
    }

    pub(crate) fn with_limits(mut self, limits: crate::SourceLimits) -> Self {
        self.limits = limits;
        self
    }

    pub(crate) fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
        self.endpoint = endpoint.into();
        self
    }

    pub(crate) fn with_lookback_messages(mut self, lookback_messages: usize) -> Self {
        self.lookback_messages = lookback_messages;
        self
    }
}

impl Source for SlackSource {
    fn name(&self) -> &str {
        "slack"
    }

    fn chunks(&self) -> Box<dyn Iterator<Item = Result<Chunk, SourceError>> + '_> {
        // `reqwest::blocking` must run off the CLI's `#[tokio::main]` thread
        // (dropping its internal runtime in an async context aborts the
        // process). Collection is eager, so run it on a scoped std thread with
        // no ambient tokio runtime.
        // Hold the scan read lease across the synchronous fetch so a
        // counter-asserting test's exclusive scope serializes this source's skip
        // recording (transport errors). A no-op in production where the gate is
        // never armed; see `skip::gate_scan`.
        crate::gate_scan(|| {
            let result = std::thread::scope(|s| match s.spawn(|| self.collect_chunks()).join() {
                Ok(result) => result,
                Err(_panic) => Err(SourceError::Other(
                    "slack fetch thread panicked".to_string(),
                )),
            });
            match result {
                Ok(chunks) => Box::new(chunks.into_iter()),
                Err(e) => Box::new(std::iter::once(Err(e))),
            }
        })
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}

#[derive(Deserialize)]
struct SlackResponse<T> {
    ok: bool,
    error: Option<String>,
    #[serde(flatten)]
    data: T,
}

#[derive(Deserialize)]
struct ConversationsList {
    channels: Option<Vec<Channel>>,
    response_metadata: Option<ResponseMetadata>,
}

#[derive(Deserialize)]
struct Channel {
    id: String,
    name: String,
}

#[derive(Deserialize)]
struct History {
    messages: Option<Vec<Message>>,
    has_more: Option<bool>,
    response_metadata: Option<ResponseMetadata>,
}

#[derive(Deserialize)]
struct ResponseMetadata {
    next_cursor: Option<String>,
}

#[derive(Deserialize)]
struct Message {
    user: Option<String>,
    text: String,
    ts: String,
}

struct ChannelListFetch {
    channels: Vec<Channel>,
    error: Option<SourceError>,
}

struct HistoryFetch {
    messages: Vec<Message>,
    error: Option<SourceError>,
}

const CONVERSATIONS_LIST: &str = "conversations.list";
const CONVERSATIONS_HISTORY: &str = "conversations.history";
const SLACK_HISTORY_PAGE_LIMIT: usize = 1000;
type SourceTruncatedReported = std::sync::atomic::AtomicBool;

fn slack_error_code(error: Option<&str>) -> &str {
    match error {
        Some(code) if !code.trim().is_empty() => code.trim(),
        _ => "<no error field>",
    }
}

fn parse_slack_response<T>(endpoint: &str, body: &str) -> Result<SlackResponse<T>, SourceError>
where
    T: DeserializeOwned,
{
    serde_json::from_str(body).map_err(|error| {
        SourceError::Other(format!(
            "failed to parse Slack {endpoint} response: {error}"
        ))
    })
}

fn read_slack_response<T>(
    endpoint: &str,
    response: Response,
    max_response_bytes: usize,
) -> Result<SlackResponse<T>, SourceError>
where
    T: DeserializeOwned,
{
    let status = response.status();
    let cap = u64::try_from(max_response_bytes).map_err(|error| {
        slack_unreadable_error(format!(
            "Slack API {endpoint} response byte cap is not addressable on this platform: {error}"
        ))
    })?;
    let capacity_hint = response
        .content_length()
        .map(|len| len.min(cap).min(64 * 1024));
    let read = crate::capped_read::read_to_cap(response, cap, capacity_hint).map_err(|error| {
        slack_unreadable_error(format!(
            "failed to read Slack {endpoint} response body: {error}"
        ))
    })?;
    if read.truncated {
        return Err(slack_response_over_cap_error(endpoint, max_response_bytes));
    }
    let body = String::from_utf8_lossy(&read.bytes);
    match parse_slack_response(endpoint, &body) {
        Ok(resp) => {
            if !status.is_success() && resp.ok {
                return Err(SourceError::Other(format!(
                    "Slack API {endpoint} returned HTTP {status} with ok=true"
                )));
            }
            Ok(resp)
        }
        Err(error) if !status.is_success() => Err(slack_unreadable_error(format!(
            "Slack API {endpoint} returned HTTP {status} and an unreadable JSON body: {error}"
        ))),
        Err(error) => Err(slack_unreadable_error(error.to_string())),
    }
}

fn slack_unreadable_error(message: String) -> SourceError {
    let _event = crate::record_skip_event(crate::SourceSkipEvent::Unreadable);
    SourceError::Other(message)
}

fn slack_response_over_cap_error(endpoint: &str, cap: usize) -> SourceError {
    let _event = crate::record_skip_event(crate::SourceSkipEvent::OverMaxSize);
    SourceError::Other(format!(
        "Slack API {endpoint} response body exceeded the {cap}-byte cap; response was not parsed and remaining Slack data was not scanned"
    ))
}

fn slack_payload_error(message: String, record_skip: bool) -> SourceError {
    if record_skip {
        slack_unreadable_error(message)
    } else {
        SourceError::Other(message)
    }
}

fn slack_next_cursor(metadata: Option<ResponseMetadata>) -> Option<String> {
    metadata
        .and_then(|metadata| metadata.next_cursor)
        .map(|cursor| cursor.trim().to_string())
        .filter(|cursor| !cursor.is_empty())
}

fn channels_page_from_response(
    resp: SlackResponse<ConversationsList>,
    record_skip: bool,
) -> Result<(Vec<Channel>, Option<String>), SourceError> {
    if !resp.ok {
        return Err(slack_payload_error(
            format!(
                "Slack API {CONVERSATIONS_LIST} error: {}",
                slack_error_code(resp.error.as_deref())
            ),
            record_skip,
        ));
    }
    let next_cursor = slack_next_cursor(resp.data.response_metadata);
    match resp.data.channels {
        Some(channels) => Ok((channels, next_cursor)),
        None => Err(slack_payload_error(
            format!("Slack API {CONVERSATIONS_LIST} ok response missing channels"),
            record_skip,
        )),
    }
}

fn channels_from_response(
    resp: SlackResponse<ConversationsList>,
) -> Result<Vec<Channel>, SourceError> {
    channels_page_from_response(resp, false).map(|(channels, _next_cursor)| channels)
}

fn messages_page_from_response(
    resp: SlackResponse<History>,
    channel_id: &str,
    record_skip: bool,
) -> Result<(Vec<Message>, Option<String>, bool), SourceError> {
    if !resp.ok {
        return Err(slack_payload_error(
            format!(
                "Slack API {CONVERSATIONS_HISTORY} error for channel {channel_id}: {}",
                slack_error_code(resp.error.as_deref())
            ),
            record_skip,
        ));
    }
    let has_more = match resp.data.has_more {
        Some(has_more) => has_more,
        None => false,
    };
    let next_cursor = slack_next_cursor(resp.data.response_metadata);
    match resp.data.messages {
        Some(messages) => Ok((messages, next_cursor, has_more)),
        None => Err(slack_payload_error(
            format!(
                "Slack API {CONVERSATIONS_HISTORY} ok response for channel {channel_id} missing messages"
            ),
            record_skip,
        )),
    }
}

fn messages_from_response(
    resp: SlackResponse<History>,
    channel_id: &str,
) -> Result<Vec<Message>, SourceError> {
    messages_page_from_response(resp, channel_id, false)
        .map(|(messages, _next_cursor, _has_more)| messages)
}

pub(crate) fn conversations_list_len_for_test(body: &str) -> Result<usize, SourceError> {
    let resp = parse_slack_response::<ConversationsList>(CONVERSATIONS_LIST, body)?;
    channels_from_response(resp).map(|channels| channels.len())
}

pub(crate) fn history_len_for_test(body: &str, channel_id: &str) -> Result<usize, SourceError> {
    let resp = parse_slack_response::<History>(CONVERSATIONS_HISTORY, body)?;
    messages_from_response(resp, channel_id).map(|messages| messages.len())
}

pub(crate) fn conversations_list_next_cursor_for_test(
    body: &str,
) -> Result<Option<String>, SourceError> {
    let resp = parse_slack_response::<ConversationsList>(CONVERSATIONS_LIST, body)?;
    channels_page_from_response(resp, false).map(|(_channels, next_cursor)| next_cursor)
}

pub(crate) fn history_next_cursor_for_test(
    body: &str,
    channel_id: &str,
) -> Result<Option<String>, SourceError> {
    let resp = parse_slack_response::<History>(CONVERSATIONS_HISTORY, body)?;
    messages_page_from_response(resp, channel_id, false)
        .map(|(_messages, next_cursor, _has_more)| next_cursor)
}

impl SlackSource {
    fn collect_chunks(&self) -> Result<Vec<Result<Chunk, SourceError>>, SourceError> {
        if self.lookback_messages == 0 {
            return Ok(Vec::new());
        }
        let http = if self.http.timeout.is_none() {
            let mut h = self.http.clone();
            h.timeout = Some(crate::timeouts::HTTP_REQUEST);
            h
        } else {
            self.http.clone()
        };
        let client = crate::http::blocking_client_builder(&http)
            .map_err(SourceError::Other)?
            .build()
            .map_err(|e| SourceError::Other(format!("failed to build Slack client: {e}")))?;

        let source_truncated_reported = SourceTruncatedReported::new(false);
        let channel_list = self.list_channels(&client, &source_truncated_reported);

        // Concurrent per-channel history fetch. Slack's tier-2 rate limit is
        // 20+ requests/minute; cap parallelism at 8 to leave headroom for the
        // burst budget. Was sequential - see the internal design notes.
        use rayon::prelude::*;
        let pool = crate::parallel_fetch::bounded_fetch_pool(
            "slack",
            crate::parallel_fetch::REMOTE_API_FETCH_THREADS,
        )?;
        let per_channel: Vec<Vec<Result<Chunk, SourceError>>> = pool.install(|| {
            channel_list
                .channels
                .par_iter()
                .map(|channel| -> Vec<Result<Chunk, SourceError>> {
                    let history =
                        self.fetch_history(&client, &channel.id, &source_truncated_reported);
                    let mut rows: Vec<Result<Chunk, SourceError>> =
                        slack_channel_chunks(channel, history.messages)
                            .into_iter()
                            .map(Ok)
                            .collect();
                    if let Some(error) = history.error {
                        rows.push(Err(error));
                    }
                    rows
                })
                .collect()
        });

        let mut chunks = Vec::new();
        for channel_chunks in per_channel {
            chunks.extend(channel_chunks);
        }
        if let Some(error) = channel_list.error {
            chunks.push(Err(error));
        }
        Ok(chunks)
    }

    fn api_url(&self, method: &str) -> String {
        format!("{}/{}", self.endpoint.trim_end_matches('/'), method)
    }

    fn list_channels(
        &self,
        client: &Client,
        source_truncated_reported: &SourceTruncatedReported,
    ) -> ChannelListFetch {
        let mut channels = Vec::new();
        let mut cursor = None::<String>;
        for _page in 1..=self.limits.hosted_git_pages {
            let mut request = client
                .get(self.api_url(CONVERSATIONS_LIST))
                .bearer_auth(&self.token)
                .query(&[
                    ("types", "public_channel,private_channel"),
                    ("limit", "1000"),
                ]);
            if let Some(cursor) = cursor.as_deref() {
                request = request.query(&[("cursor", cursor)]);
            }
            let resp = match request.send() {
                Ok(resp) => resp,
                Err(error) => {
                    return ChannelListFetch {
                        channels,
                        error: Some(slack_unreadable_error(format!(
                            "Slack API {CONVERSATIONS_LIST} request failed: {error}"
                        ))),
                    };
                }
            };
            let resp =
                match read_slack_response(CONVERSATIONS_LIST, resp, self.limits.web_response_bytes)
                {
                    Ok(resp) => resp,
                    Err(error) => {
                        return ChannelListFetch {
                            channels,
                            error: Some(error),
                        };
                    }
                };
            let (page_channels, next_cursor) = match channels_page_from_response(resp, true) {
                Ok(page) => page,
                Err(error) => {
                    return ChannelListFetch {
                        channels,
                        error: Some(error),
                    };
                }
            };
            channels.extend(page_channels);
            let Some(next_cursor) = next_cursor else {
                return ChannelListFetch {
                    channels,
                    error: None,
                };
            };
            cursor = Some(next_cursor);
        }
        ChannelListFetch {
            channels,
            error: Some(slack_truncated_error(
                CONVERSATIONS_LIST,
                "channel listing",
                self.limits.hosted_git_pages,
                source_truncated_reported,
            )),
        }
    }

    fn fetch_history(
        &self,
        client: &Client,
        channel_id: &str,
        source_truncated_reported: &SourceTruncatedReported,
    ) -> HistoryFetch {
        let mut messages = Vec::new();
        let mut cursor = None::<String>;
        for _page in 1..=self.limits.hosted_git_pages {
            let remaining = self.lookback_messages.saturating_sub(messages.len());
            let limit = remaining.min(SLACK_HISTORY_PAGE_LIMIT).to_string();
            let mut request = client
                .get(self.api_url(CONVERSATIONS_HISTORY))
                .bearer_auth(&self.token)
                .query(&[("channel", channel_id), ("limit", &limit)]);
            if let Some(cursor) = cursor.as_deref() {
                request = request.query(&[("cursor", cursor)]);
            }
            let resp = match request.send() {
                Ok(resp) => resp,
                Err(error) => {
                    return HistoryFetch {
                        messages,
                        error: Some(slack_unreadable_error(format!(
                    "Slack API {CONVERSATIONS_HISTORY} request failed for channel {channel_id}: {error}"
                ))),
                    };
                }
            };
            let resp = match read_slack_response(
                CONVERSATIONS_HISTORY,
                resp,
                self.limits.web_response_bytes,
            ) {
                Ok(resp) => resp,
                Err(error) => {
                    return HistoryFetch {
                        messages,
                        error: Some(error),
                    };
                }
            };
            let (page_messages, next_cursor, has_more) =
                match messages_page_from_response(resp, channel_id, true) {
                    Ok(page) => page,
                    Err(error) => {
                        return HistoryFetch {
                            messages,
                            error: Some(error),
                        };
                    }
                };
            let page_exceeded_remaining = page_messages.len() > remaining;
            messages.extend(page_messages.into_iter().take(remaining));
            if messages.len() >= self.lookback_messages {
                let error = if has_more || page_exceeded_remaining {
                    Some(slack_lookback_truncated_error(
                        CONVERSATIONS_HISTORY,
                        &format!("history for channel {channel_id}"),
                        self.lookback_messages,
                        source_truncated_reported,
                    ))
                } else {
                    None
                };
                return HistoryFetch { messages, error };
            }
            match (next_cursor, has_more) {
                (Some(next_cursor), true) => cursor = Some(next_cursor),
                (Some(_), false) | (None, false) => {
                    return HistoryFetch {
                        messages,
                        error: None,
                    };
                }
                (None, true) => {
                    return HistoryFetch {
                        messages,
                        error: Some(slack_missing_cursor_error(
                            CONVERSATIONS_HISTORY,
                            &format!("history for channel {channel_id}"),
                            source_truncated_reported,
                        )),
                    };
                }
            }
        }
        HistoryFetch {
            messages,
            error: Some(slack_truncated_error(
                CONVERSATIONS_HISTORY,
                &format!("history for channel {channel_id}"),
                self.limits.hosted_git_pages,
                source_truncated_reported,
            )),
        }
    }
}

fn slack_channel_chunks(channel: &Channel, messages: Vec<Message>) -> Vec<Chunk> {
    let mut channel_chunks = Vec::new();
    let mut channel_buffer = String::new();
    for msg in messages {
        if let Some(user) = &msg.user {
            channel_buffer.push_str(&format!("\n[USER: {} TS: {}]\n", user, msg.ts));
        }
        channel_buffer.push_str(&msg.text);
        channel_buffer.push('\n');
        if channel_buffer.len() > 64 * 1024 {
            channel_chunks.push(Chunk {
                data: std::mem::take(&mut channel_buffer).into(),
                metadata: ChunkMetadata {
                    base_offset: 0,
                    base_line: 0,
                    source_type: "slack".into(),
                    path: Some(format!("slack://#{}", channel.name).into()),
                    ..Default::default()
                },
            });
        }
    }
    if !channel_buffer.is_empty() {
        channel_chunks.push(Chunk {
            data: channel_buffer.into(),
            metadata: ChunkMetadata {
                base_offset: 0,
                base_line: 0,
                source_type: "slack".into(),
                path: Some(format!("slack://#{}", channel.name).into()),
                ..Default::default()
            },
        });
    }
    channel_chunks
}

fn record_slack_source_truncated_once(source_truncated_reported: &SourceTruncatedReported) {
    if !source_truncated_reported.swap(true, std::sync::atomic::Ordering::Relaxed) {
        let _event = crate::record_skip_event(crate::SourceSkipEvent::SourceTruncated);
    }
}

fn slack_truncated_error(
    endpoint: &str,
    item: &str,
    max_pages: usize,
    source_truncated_reported: &SourceTruncatedReported,
) -> SourceError {
    record_slack_source_truncated_once(source_truncated_reported);
    SourceError::Other(format!(
        "Slack API {endpoint} {item} exceeded {max_pages} pages; remaining Slack data was not scanned"
    ))
}

fn slack_missing_cursor_error(
    endpoint: &str,
    item: &str,
    source_truncated_reported: &SourceTruncatedReported,
) -> SourceError {
    record_slack_source_truncated_once(source_truncated_reported);
    SourceError::Other(format!(
        "Slack API {endpoint} {item} indicated more pages without a next cursor; remaining Slack data was not scanned"
    ))
}

fn slack_lookback_truncated_error(
    endpoint: &str,
    item: &str,
    lookback_messages: usize,
    source_truncated_reported: &SourceTruncatedReported,
) -> SourceError {
    record_slack_source_truncated_once(source_truncated_reported);
    SourceError::Other(format!(
        "Slack API {endpoint} {item} reached the {lookback_messages}-message lookback cap; remaining Slack data was not scanned"
    ))
}