mirror of
https://github.com/tux3/crowdstrike-cloudproto
synced 2026-06-08 17:59:21 +00:00
TsEventSocket carries Events over a CloudProtoSocket
This allows connecting to a TS server and exchanging packets with it Note that we don't implement listenning on this socket yet, so for now it can only be used to build a client.
This commit is contained in:
tux3
+2
-2
@@ -11,12 +11,12 @@ bytes = "1.2.1"
|
||||
byteorder = "1.4.3"
|
||||
thiserror = "1.0.32"
|
||||
tracing = "0.1.36"
|
||||
strum = "0.24.1"
|
||||
strum_macros = "0.24.3"
|
||||
hex = "0.4.3"
|
||||
|
||||
[dev-dependencies]
|
||||
tokio = { version = "1", features = ["io-util", "macros", "rt-multi-thread"] }
|
||||
strum = "0.24.1"
|
||||
strum_macros = "0.24.3"
|
||||
rand = "0.8.5"
|
||||
anyhow = "1.0.62"
|
||||
test-log = { version = "0.2.11", features = ["trace"], default-features = false }
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
Provides async sockets and high-level objects that implement the protocol
|
||||
used between Crowdstrike's Falcon Sensor and the two backend services:
|
||||
- The TS event server (event collection, device monitoring, admin remote shell, ...)
|
||||
- The LFO file server (downloading updates, uploading sample files for analysis, ...)
|
||||
|
||||
It is also possible to implement your own private TS or LFO server,
|
||||
for instance if you want to receive live sensor events from the official client.
|
||||
|
||||
## Features
|
||||
|
||||
The [`CloudProtoSocket`](framing::CloudProtoSocket) implements the common low-level packet structure
|
||||
used by the official client and both cloud services.
|
||||
You probably want to use the higher-level TS socket or LFO client directly;
|
||||
They are both layerd over a [`CloudProtoSocket`](framing::CloudProtoSocket),
|
||||
but they work with high level concepts instead of [`CloudProtoPacket`](framing::CloudProtoPacket)s.
|
||||
|
||||
### TS Event socket
|
||||
|
||||
The [`TsEventSocket`](services::ts::TsEventSocket) allows connecting to the TS service
|
||||
and exchanging [`Event`](services::ts::Event)s,
|
||||
which is how the `falcon-sensor` agent streams back live information to the cloud.
|
||||
|
||||
You must provide a valid Customer ID (CID) to connect to the official TS servers.
|
||||
See the [`TsEventSocket`](services::ts::TsEventSocket) documentation for more information.
|
||||
|
||||
### LFO client
|
||||
|
||||
> The LFO client is **not published yet**. Coming Soon™.
|
||||
|
||||
The `LfoClient` allows you to download updates and other potentially large files used by the sensor.
|
||||
|
||||
The client supports LFO chunked downloads and optional XZ compression.
|
||||
There is currently no immediate plan to support uploads.
|
||||
|
||||
You do not need to be a Crowdstrike customer to download files from LFO.
|
||||
(LFO requests contain CID/AID fields, but any values are accepted).
|
||||
|
||||
### Server functionality
|
||||
|
||||
Running a third party Crowdstrike server requires using a modified client configured
|
||||
to connect to a domain you own with a valid certificate,
|
||||
or disabling certificate validation in falcon-sensor.
|
||||
|
||||
As of version 13601, Falcon as a whole performs no integrity checks, so it happily runs with arbitrary patches applied.
|
||||
|
||||
### Epistemic Notice
|
||||
|
||||
Please note that this crate is a clean-room implementation based on observing sensor version 13601
|
||||
talk to a third-party server in an isolated VM,
|
||||
and on using this crate to replay a few sensor events and capturing the public TS service's replies.
|
||||
|
||||
As a result, you should expect that this library **may not be a 100% conforming implementation**.
|
||||
It may be missing some optional parts of the protocol, some of the reverse-engineered fields that
|
||||
don't affect the result may lack names, and some might be named wrong entirely.
|
||||
|
||||
## What is the Crowdstrike CLOUDPROTO?
|
||||
|
||||
The name "CLOUDPROTO" comes from a debug log message inside the falcon-sensor binary.
|
||||
|
||||
Internally, falcon-sensor is architectured around Actors (C++ objects) that exchange
|
||||
Events (plain data) using an event bus.
|
||||
The same events that falcon-sensor actors use internally are also used to communicate with the TS cloud server
|
||||
by carrying serialized [events](services::ts::Event) within CLOUDPROTO [frames](framing::CloudProtoPacket).
|
||||
|
||||
Events for the TS server are first serialized with Protobuf and complemented by
|
||||
a short header that contains the event type and a transaction ID.
|
||||
This serialized event payload is sent over a CLOUDPROTO socket, which is itself wrapped
|
||||
in a TLS session over TCP port 443.
|
||||
|
||||
The event protocol has a quirky ACK mechanism, which appears redundant with the TLS and TCP sockets
|
||||
it's layered over, and does not in fact seem to be actually used to provide any backpressure
|
||||
or retransmission guarantees in the official implementation.
|
||||
The `falcon-sensor` does send ACK packets, but seems to ignore incoming ACKs (or their lack of) entirely.
|
||||
(This quirk is despite much of the functionality to track ACKs and in-flight packets
|
||||
being visibly present in the `falcon-sensor` binary).
|
||||
|
||||
The LFO server also uses CLOUDPROTO frames to carry its messages,
|
||||
but instead of TS events it uses simple request/response packets,
|
||||
and has no ACK mechanism at all.
|
||||
+11
-7
@@ -5,24 +5,28 @@
|
||||
//! but ignores the inner service-specific payload format and interpretation of packet kinds.
|
||||
|
||||
mod hdr_version;
|
||||
pub mod packet;
|
||||
mod packet;
|
||||
mod socket;
|
||||
|
||||
pub use hdr_version::CloudProtoVersion;
|
||||
pub use packet::CloudProtoPacket;
|
||||
pub use socket::CloudProtoSocket;
|
||||
|
||||
use crate::services::CloudProtoMagic;
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Error, Debug)]
|
||||
pub enum CloudProtoError {
|
||||
#[error("Bad CloudProto magic {0:#x}, expected {1:#x}")]
|
||||
BadMagic(u8, u8),
|
||||
BadMagic(CloudProtoMagic, CloudProtoMagic),
|
||||
#[error("Bad CloudProto header version {0:#x}, expected {1:#x}")]
|
||||
BadVersion(u16, u16),
|
||||
#[error("Bad CloudProto payload size {0:#x}, header announced {1:#x}")]
|
||||
BadSize(usize, usize),
|
||||
#[error("Received packet kind {0} as connection reply, expected {1}")]
|
||||
WrongConnectionEstablishedKind(u8, u8),
|
||||
BadVersion(CloudProtoVersion, CloudProtoVersion),
|
||||
#[error("Bad CloudProto payload size {0:#x}, frame header announced {1:#x}")]
|
||||
BadFrameSize(usize, usize),
|
||||
#[error("Received payload size too short, got {0:#x} but wanted at least {1:#x}")]
|
||||
PayloadTooShort(usize, usize),
|
||||
#[error("Received packet kind {0} while connecting, but expected {1}")]
|
||||
WrongConnectionPacketKind(u8, u8),
|
||||
#[error("{0}")]
|
||||
ClosedByPeer(String),
|
||||
#[error("CloudProto IO error")]
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
use strum_macros::{Display, EnumCount, FromRepr};
|
||||
|
||||
#[repr(u16)]
|
||||
#[derive(Eq, PartialEq, Copy, Clone, Debug)]
|
||||
#[cfg_attr(test, derive(strum_macros::EnumCount))]
|
||||
#[derive(Eq, PartialEq, Copy, Clone, Debug, Display, EnumCount, FromRepr)]
|
||||
pub enum CloudProtoVersion {
|
||||
/// All packets that don't fall in other categories
|
||||
Normal,
|
||||
@@ -36,6 +37,20 @@ impl From<&CloudProtoVersion> for u16 {
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::LowerHex for CloudProtoVersion {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
let val: u16 = self.into();
|
||||
std::fmt::LowerHex::fmt(&val, f)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::UpperHex for CloudProtoVersion {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
let val: u16 = self.into();
|
||||
std::fmt::UpperHex::fmt(&val, f)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod test {
|
||||
use crate::framing::CloudProtoVersion;
|
||||
|
||||
@@ -13,7 +13,6 @@ pub struct CloudProtoPacket {
|
||||
/// Each value can have a different interpretation for each backend service
|
||||
/// There is no common definition of packet kind at the framing level
|
||||
pub kind: u8,
|
||||
/// Used
|
||||
pub version: CloudProtoVersion,
|
||||
pub payload: Vec<u8>,
|
||||
}
|
||||
@@ -27,7 +26,7 @@ impl CloudProtoPacket {
|
||||
let pkt_size = reader.read_u32::<BE>()? as usize - COMMON_HDR_LEN;
|
||||
let remaining_size = buf.len() - reader.position() as usize;
|
||||
if remaining_size != pkt_size {
|
||||
return Err(CloudProtoError::BadSize(remaining_size, pkt_size));
|
||||
return Err(CloudProtoError::BadFrameSize(remaining_size, pkt_size));
|
||||
}
|
||||
let payload = buf[reader.position() as usize..].to_vec();
|
||||
Ok(Self {
|
||||
|
||||
+8
-11
@@ -1,5 +1,4 @@
|
||||
use crate::framing::packet::CloudProtoPacket;
|
||||
use crate::framing::CloudProtoError;
|
||||
use bytes::Bytes;
|
||||
use futures_util::{Sink, SinkExt, Stream, StreamExt};
|
||||
use std::pin::Pin;
|
||||
@@ -15,9 +14,9 @@ pub struct CloudProtoSocket<IO> {
|
||||
|
||||
impl<IO> CloudProtoSocket<IO>
|
||||
where
|
||||
IO: AsyncRead + AsyncWrite + Unpin,
|
||||
IO: AsyncRead + AsyncWrite,
|
||||
{
|
||||
pub async fn new(io: IO) -> Result<Self, CloudProtoError> {
|
||||
pub fn new(io: IO) -> Self {
|
||||
let (read, write) = tokio::io::split(io);
|
||||
let read = LengthDelimitedCodec::builder()
|
||||
.big_endian()
|
||||
@@ -27,13 +26,13 @@ where
|
||||
.num_skip(0)
|
||||
.new_read(read);
|
||||
let write = FramedWrite::new(write, BytesCodec::new());
|
||||
Ok(Self { read, write })
|
||||
Self { read, write }
|
||||
}
|
||||
}
|
||||
|
||||
impl<IO> Stream for CloudProtoSocket<IO>
|
||||
where
|
||||
IO: AsyncRead + AsyncWrite + Unpin,
|
||||
IO: AsyncRead + AsyncWrite,
|
||||
{
|
||||
type Item = CloudProtoPacket;
|
||||
|
||||
@@ -69,7 +68,7 @@ where
|
||||
|
||||
impl<IO> Sink<CloudProtoPacket> for CloudProtoSocket<IO>
|
||||
where
|
||||
IO: AsyncRead + AsyncWrite + Unpin,
|
||||
IO: AsyncRead + AsyncWrite,
|
||||
{
|
||||
type Error = std::io::Error;
|
||||
|
||||
@@ -100,10 +99,8 @@ where
|
||||
|
||||
#[cfg(test)]
|
||||
mod test {
|
||||
use crate::framing::packet::CloudProtoPacket;
|
||||
use crate::framing::CloudProtoVersion;
|
||||
use crate::framing::{CloudProtoPacket, CloudProtoSocket, CloudProtoVersion};
|
||||
use crate::services::CloudProtoMagic;
|
||||
use crate::CloudProtoSocket;
|
||||
use anyhow::Result;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use rand::Rng;
|
||||
@@ -111,8 +108,8 @@ mod test {
|
||||
#[test_log::test(tokio::test)]
|
||||
async fn single_send_recv() -> Result<()> {
|
||||
let (client, server) = tokio::io::duplex(100 * 1024);
|
||||
let mut client = CloudProtoSocket::new(client).await?;
|
||||
let mut server = CloudProtoSocket::new(server).await?;
|
||||
let mut client = CloudProtoSocket::new(client);
|
||||
let mut server = CloudProtoSocket::new(server);
|
||||
|
||||
let mut rng = rand::thread_rng();
|
||||
let len = rng.gen::<u16>() as usize;
|
||||
|
||||
+1
-43
@@ -1,46 +1,4 @@
|
||||
//! The binary protocol between Crowdstrike's Falcon Sensor and cloud servers
|
||||
//!
|
||||
//! Provides an async socket layer and structure definitions for the internal binary protocol used between
|
||||
//! falcon-sensor and its backend cloud services.
|
||||
//!
|
||||
//! There are two Crowdstrike services you can talk to using this protocol, both over TLS:
|
||||
//! - The TS event server (event collection, device monitoring, admin remote shell, ...)
|
||||
//! - The LFO file server (downloading updates, uploading sample files for analysis, ...)
|
||||
//!
|
||||
//! You can also use this crate to act as a private TS (or LFO) server that receives and decodes live sensor events.
|
||||
//! This requires disabling certificate validation in falcon-sensor, although as of version 13601
|
||||
//! Falcon as a whole does no integrity checks, so that it happily runs with arbitrary patches applied.
|
||||
//!
|
||||
//! ## Notice
|
||||
//!
|
||||
//! Please note that this crate is a clean-room implementation based on observing sensor version 13601 talk to a private server in an isolated VM,
|
||||
//! and on using this crate to replay a few sensor events and capturing the public TS service's replies.
|
||||
//!
|
||||
//! As a result, you should expect that this library **may not be a 100% conforming implementation**.
|
||||
//! It may be missing some optional parts of the protocol, some of the reverse-engineered fields that
|
||||
//! don't affect the result may lack names, and some might be named wrong entirely.
|
||||
//!
|
||||
//! ## What is the Crowdstrike CLOUDPROTO?
|
||||
//!
|
||||
//! The name "CLOUDPROTO" comes from a debug log message inside the falcon-sensor binary.
|
||||
//!
|
||||
//! Internally, falcon-sensor is architectured around Actors (C++ objects) that exchange Events (plain data) using an event bus.
|
||||
//! The same events that falcon-sensor actors use internally are also used to communicate
|
||||
//! with the TS cloud server by carrying serialized events within CLOUDPROTO frames.
|
||||
//!
|
||||
//! Events are first serialized with Protobuf and complemented by a short header that contains
|
||||
//! the event type and a transaction ID.
|
||||
//! This serialized event payload is sent over a CLOUDPROTO socket, which is itself wrapped
|
||||
//! in a TLS session over TCP port 443.
|
||||
//!
|
||||
//! This crate provides the CLOUDPROTO socket layer that handles framing and the outer header,
|
||||
//! and a higher level Event socket that takes care of ACK'ing cloud proto frames and turning them
|
||||
//! into bare Event structs that contain the event type and the serialized Protobuf payload.
|
||||
//!
|
||||
//! We provide an EventType enum that tries to give a name to a few common events,
|
||||
//! however the Protobuf schemas corresponding to the many types of event payloads are not part of this library.
|
||||
#![doc = include_str!("../README.md")]
|
||||
|
||||
pub mod framing;
|
||||
pub mod services;
|
||||
|
||||
pub use framing::CloudProtoSocket;
|
||||
|
||||
+19
-7
@@ -1,13 +1,12 @@
|
||||
mod lfo;
|
||||
mod ts;
|
||||
pub mod lfo;
|
||||
pub mod ts;
|
||||
|
||||
pub use lfo::*;
|
||||
pub use ts::*;
|
||||
use strum_macros::{Display, EnumCount, FromRepr};
|
||||
|
||||
/// This CID is **NOT** structurally valid, it would not be accepted by the sensor.
|
||||
/// It is also possible to use a structurally valid CID that belongs to no one, but all zeros are accepted by LFO.
|
||||
pub const DEFAULT_CID_HEX: &str = "00000000000000000000000000000000";
|
||||
/// The AID is a value assigned to clients by TS on connection. It's zero for new agents.
|
||||
/// The AID is a value assigned to clients by TS on connection. It is zero for new agents.
|
||||
pub const DEFAULT_AID_HEX: &str = "00000000000000000000000000000000";
|
||||
/// Arbitrary machine-specific value generated on an isolated VM (from /proc/sys/kernel/random/boot_id)
|
||||
pub const DEFAULT_BOOTID_HEX: &str = "6c959680d4945d45924301a720debc88";
|
||||
@@ -15,8 +14,7 @@ pub const DEFAULT_BOOTID_HEX: &str = "6c959680d4945d45924301a720debc88";
|
||||
pub const DEFAULT_UNK0_HEX: &str = "54645dacc392cb43b4803094141e0087";
|
||||
|
||||
#[repr(u8)]
|
||||
#[derive(Eq, PartialEq, Debug, Copy, Clone)]
|
||||
#[cfg_attr(test, derive(strum_macros::EnumCount))]
|
||||
#[derive(Eq, PartialEq, Debug, Copy, Clone, Display, EnumCount, FromRepr)]
|
||||
pub enum CloudProtoMagic {
|
||||
TS,
|
||||
LFO,
|
||||
@@ -61,6 +59,20 @@ impl PartialEq<CloudProtoMagic> for u8 {
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::LowerHex for CloudProtoMagic {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
let val: u8 = self.into();
|
||||
std::fmt::LowerHex::fmt(&val, f)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::UpperHex for CloudProtoMagic {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
let val: u8 = self.into();
|
||||
std::fmt::UpperHex::fmt(&val, f)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod test {
|
||||
use super::CloudProtoMagic;
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
use strum_macros::{Display, EnumCount, FromRepr};
|
||||
|
||||
/// Besides transporting events, the protocol carries different kind of packets internally
|
||||
#[repr(u8)]
|
||||
#[derive(Eq, PartialEq, Copy, Clone, Debug)]
|
||||
#[cfg_attr(test, derive(strum_macros::EnumCount))]
|
||||
#[derive(Eq, PartialEq, Copy, Clone, Debug, Display, EnumCount, FromRepr)]
|
||||
pub enum LfoPacketKind {
|
||||
/// Sent by client to request file data
|
||||
GetFileRequest,
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
mod event;
|
||||
mod pkt_kind;
|
||||
mod socket;
|
||||
|
||||
pub use event::{Event, EventId};
|
||||
pub use pkt_kind::TsPacketKind;
|
||||
pub use socket::TsEventSocket;
|
||||
|
||||
use crate::services::{DEFAULT_BOOTID_HEX, DEFAULT_UNK0_HEX};
|
||||
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
use crate::framing::CloudProtoError;
|
||||
use byteorder::{ReadBytesExt, WriteBytesExt, BE};
|
||||
use std::io::{Read, Write};
|
||||
use strum_macros::{AsRefStr, Display, FromRepr};
|
||||
|
||||
// Does not count the txid, which is handled transparently in the TsEventSocket
|
||||
pub(crate) const EVT_HDR_LEN: usize = 4;
|
||||
|
||||
/// The `data` field usually contains a serialized Protobuf structure.
|
||||
///
|
||||
/// The Protobuf schema of the `data` depends entirely on `raw_event_id`,
|
||||
/// and this crate does not provide deserialization for specific events.
|
||||
///
|
||||
/// A few event IDs do not correspond to protobuf data at all, using a variety of other simple binary formats.
|
||||
///
|
||||
/// The `event_id` field is `None` for values of `raw_event_id` that are not in the [`EventId`](EventId) enum.
|
||||
#[derive(Eq, PartialEq, Debug, Clone)]
|
||||
pub struct Event {
|
||||
pub raw_event_id: u32,
|
||||
pub event_id: Option<EventId>,
|
||||
pub data: Vec<u8>,
|
||||
}
|
||||
|
||||
impl Event {
|
||||
pub fn new(event_id: EventId, data: Vec<u8>) -> Self {
|
||||
Self {
|
||||
raw_event_id: event_id as u32,
|
||||
event_id: Some(event_id),
|
||||
data,
|
||||
}
|
||||
}
|
||||
|
||||
/// Best effort text representation of the event ID using know [`EventId`][EventId] values
|
||||
pub fn ev_id_string(&self) -> String {
|
||||
if let Some(id) = self.event_id {
|
||||
id.to_string()
|
||||
} else {
|
||||
self.raw_event_id.to_string()
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn from_read(reader: &mut dyn Read) -> Result<Self, CloudProtoError> {
|
||||
let raw_event_id = reader.read_u32::<BE>()?;
|
||||
let event_id = EventId::from_repr(raw_event_id);
|
||||
let mut data = Vec::new();
|
||||
reader.read_to_end(&mut data)?;
|
||||
Ok(Self {
|
||||
raw_event_id,
|
||||
event_id,
|
||||
data,
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn into_write(self, writer: &mut dyn Write) -> Result<(), CloudProtoError> {
|
||||
writer.write_u32::<BE>(self.raw_event_id)?;
|
||||
writer.write_all(&self.data)?;
|
||||
writer.flush()?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Tries to provide meaningful names for some well-known [`Event`](Event)s.
|
||||
///
|
||||
/// Event ID names containing `UNK` are known to exist, but were not found in the form of an immediate value in falcon-sensor,
|
||||
/// so that some effort (beyond a simple lookup in the client) may be required to name them.
|
||||
///
|
||||
/// Some Event IDs are internal to falcon-sensor and never leave on the wire, so they are not listed.
|
||||
/// Some events have not been observed yet, or may be added in later updates,
|
||||
/// so this enum is only meant to document well-known values as a best-effort.
|
||||
/// It won't be an exhaustive list.
|
||||
#[derive(Eq, PartialEq, Debug, Copy, Clone, Display, AsRefStr, FromRepr)]
|
||||
#[repr(u32)]
|
||||
#[rustfmt::skip]
|
||||
#[allow(non_camel_case_types)]
|
||||
#[allow(dead_code)]
|
||||
#[non_exhaustive]
|
||||
pub enum EventId {
|
||||
ConfigurationLoaded = 0x308000AA,
|
||||
LfoDownloadFromManifestRecord = 0x308000AD,
|
||||
UNK_SERVER_0x30800207 = 0x30800207, // Sent by server, but no search results in sensor
|
||||
CurrentSystemTags = 0x30800208,
|
||||
CloudRequestReceived = 0x3080028E,
|
||||
UNK_0x30800296 = 0x30800296, // No search results in sensor
|
||||
IpAddressAddedForFamily2 = 0x308002e5, // I think this is IPv4, and the other IPv6?
|
||||
IpAddressAdded = 0x308002e6,
|
||||
HostnameChanged = 0x3080034D,
|
||||
CurrentUninstallTokenInfo = 0x30800457,
|
||||
DiskCapacity = 0x3080069f,
|
||||
DiskUtilization = 0x30800850,
|
||||
UNK_0x31000002 = 0x31000002, // No search results in sensor
|
||||
ChannelVersionRequired = 0x310001D1,
|
||||
UNK_0x3100053f = 0x3100053f, // No search results in sensor
|
||||
SystemCapacity = 0x310005AB,
|
||||
UpdateCloudEvent = 0x318002B1,
|
||||
OsVersionInfo = 0x3200014e,
|
||||
UNK_0x32000220 = 0x32000220, // No search results in sensor
|
||||
UNK_0x320002cf = 0x320002cf, // No search results in sensor
|
||||
ConnectionStatus = 0x32800139,
|
||||
AgentOnline = 0x338000ac,
|
||||
UNK_0x340000ee = 0x340000ee, // No search results in sensor
|
||||
}
|
||||
@@ -1,7 +1,8 @@
|
||||
use strum_macros::{Display, EnumCount, FromRepr};
|
||||
|
||||
/// Besides transporting events, the TS sub-protocol has handshake packets and an ACK mechanism
|
||||
#[repr(u8)]
|
||||
#[derive(Eq, PartialEq, Copy, Clone, Debug)]
|
||||
#[cfg_attr(test, derive(strum_macros::EnumCount))]
|
||||
#[derive(Eq, PartialEq, Copy, Clone, Debug, Display, EnumCount, FromRepr)]
|
||||
pub enum TsPacketKind {
|
||||
/// First packet from client to server
|
||||
Connect,
|
||||
@@ -10,8 +11,8 @@ pub enum TsPacketKind {
|
||||
/// Application data is carried with this packet kind
|
||||
/// Usually contains Event messages with a tx id (for ACKs) and other fields depending on the event's Protobuf schema.
|
||||
Event,
|
||||
/// CloudProto is normally carried over TLS, but it still has ACKs (seemingly to enforce backpressure).
|
||||
/// Too many event messages sent without waiting for ACKs will be dropped by the other side.
|
||||
/// CloudProto is normally carried over TLS, but can still use an ACK mechanism.
|
||||
/// In practice the official client largely ignores ACKs, and we try to follow its behavior.
|
||||
Ack,
|
||||
/// This escape hatch is provided with no warranty including fitness for a particular purpose.
|
||||
/// Good luck!
|
||||
|
||||
@@ -0,0 +1,346 @@
|
||||
use crate::framing::{CloudProtoError, CloudProtoPacket, CloudProtoSocket, CloudProtoVersion};
|
||||
use crate::services::ts::event::EVT_HDR_LEN;
|
||||
use crate::services::ts::{Event, TsConnectInfo, TsPacketKind};
|
||||
use crate::services::CloudProtoMagic;
|
||||
use futures_util::{Sink, SinkExt, Stream, StreamExt};
|
||||
use std::io::Cursor;
|
||||
use std::pin::Pin;
|
||||
use std::task::{ready, Context, Poll};
|
||||
use tokio::io::{AsyncRead, AsyncWrite};
|
||||
use tracing::{debug, error, trace, warn};
|
||||
|
||||
const HDR_TXID_SIZE: usize = std::mem::size_of::<u64>();
|
||||
// Values observed from the official client.
|
||||
// The TS server returns large quickly incrementing TXIDs, but these values here are fine.
|
||||
const FIRST_TXID: u64 = 0x200;
|
||||
const TXID_INCREMENT: u64 = 0x100;
|
||||
|
||||
#[repr(u8)]
|
||||
enum AgentIdStatus {
|
||||
Unchanged = 0x1,
|
||||
Changed = 0x2,
|
||||
}
|
||||
|
||||
/// Async socket used to stream [`Event`](Event)s with the TS service
|
||||
///
|
||||
/// You need to provide a valid Crowdstrike Customer ID (CID) to authenticate with the server.
|
||||
/// The TS server checks that this CID belongs to a valid customer and will immediately close the socket otherwise.
|
||||
///
|
||||
/// You should have been provided with a "CCID" when installing the Falcon Sensor,
|
||||
/// which looks something like "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA-BB".
|
||||
/// The CID is the first part before the "-BB".
|
||||
///
|
||||
/// After installation, you can still find your CID in binary form in the "falconstore" file,
|
||||
/// saved as a 16 byte binary blob, right after the UTF-16 literal "CU".
|
||||
pub struct TsEventSocket<IO> {
|
||||
io: CloudProtoSocket<IO>,
|
||||
next_txid: u64,
|
||||
|
||||
unacked_txid: Option<u64>,
|
||||
unacked_event: Option<Event>,
|
||||
}
|
||||
|
||||
impl<IO> TsEventSocket<IO>
|
||||
where
|
||||
IO: AsyncRead + AsyncWrite,
|
||||
{
|
||||
pub async fn connect(
|
||||
mut io: CloudProtoSocket<IO>,
|
||||
info: TsConnectInfo,
|
||||
) -> Result<Self, CloudProtoError> {
|
||||
let mut payload = Vec::with_capacity(4 * 16 + 8);
|
||||
payload.extend_from_slice(&info.cid);
|
||||
payload.extend_from_slice(&info.unk0);
|
||||
payload.extend_from_slice(&info.aid);
|
||||
payload.extend_from_slice(&info.bootid);
|
||||
payload.extend_from_slice(&info.pt);
|
||||
let pkt = CloudProtoPacket {
|
||||
magic: CloudProtoMagic::TS,
|
||||
kind: TsPacketKind::Connect.into(),
|
||||
version: CloudProtoVersion::Connect,
|
||||
payload,
|
||||
};
|
||||
io.send(pkt).await?;
|
||||
|
||||
let reply = match io.next().await {
|
||||
Some(pkt) => pkt,
|
||||
None => {
|
||||
return Err(CloudProtoError::ClosedByPeer(
|
||||
"TS server closed connection".into(),
|
||||
))
|
||||
}
|
||||
};
|
||||
// Log the connection packet for debugging, since we don't otherwise return the payload in errors
|
||||
trace!("Received TS connect reply: {}", hex::encode(&reply.payload));
|
||||
|
||||
if reply.magic != CloudProtoMagic::TS {
|
||||
return Err(CloudProtoError::BadMagic(reply.magic, CloudProtoMagic::TS));
|
||||
}
|
||||
if reply.kind != TsPacketKind::ConnectionEstablished {
|
||||
error!(
|
||||
"Bad TS connect reply kind: {:X?}, payload: {}",
|
||||
reply,
|
||||
hex::encode(&reply.payload)
|
||||
);
|
||||
return Err(CloudProtoError::WrongConnectionPacketKind(
|
||||
reply.kind,
|
||||
TsPacketKind::ConnectionEstablished.into(),
|
||||
));
|
||||
}
|
||||
if reply.version != CloudProtoVersion::Normal {
|
||||
error!(
|
||||
"Bad TS connect reply version: {:X?}, payload: {}",
|
||||
reply,
|
||||
hex::encode(&reply.payload)
|
||||
);
|
||||
return Err(CloudProtoError::BadVersion(
|
||||
reply.version,
|
||||
CloudProtoVersion::Normal,
|
||||
));
|
||||
}
|
||||
|
||||
if reply.payload.len() != 17 {
|
||||
warn!("TsEventSocket connect reply has unexpected size, continuing anyways")
|
||||
} else if reply.payload[0] == AgentIdStatus::Unchanged as u8 {
|
||||
debug!(
|
||||
received_aid = hex::encode(&reply.payload[1..]),
|
||||
"TS socket connected, AgentID unchanged",
|
||||
);
|
||||
if info.aid[..] != reply.payload {
|
||||
warn!("TS server says to keep our AgentID, but replied with a different one!");
|
||||
}
|
||||
} else if reply.payload[0] == AgentIdStatus::Changed as u8 {
|
||||
debug!(
|
||||
received_aid = hex::encode(&reply.payload[1..]),
|
||||
"TS socket connected, AgentID has changed",
|
||||
);
|
||||
if info.aid[..] == reply.payload {
|
||||
warn!("TS server says to change our AgentID, but replied with the same one!");
|
||||
}
|
||||
} else {
|
||||
warn!(
|
||||
"Unexpected value from TS server when checking whether the AgentID changed: {:#x}",
|
||||
reply.payload[0]
|
||||
)
|
||||
}
|
||||
|
||||
Ok(Self {
|
||||
io,
|
||||
next_txid: FIRST_TXID,
|
||||
unacked_txid: None,
|
||||
unacked_event: None,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl<IO> Stream for TsEventSocket<IO>
|
||||
where
|
||||
IO: AsyncRead + AsyncWrite,
|
||||
{
|
||||
type Item = Result<Event, CloudProtoError>;
|
||||
|
||||
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
let this = self.get_mut();
|
||||
|
||||
// (Shh, don't tell anyone, but this is a stealth goto we take just once after receiving an event!)
|
||||
'process_pending_acks: loop {
|
||||
if let Some(txid) = &this.unacked_txid {
|
||||
assert!(this.unacked_event.is_some());
|
||||
ready!(this.io.poll_ready_unpin(cx))?;
|
||||
|
||||
this.io.start_send_unpin(CloudProtoPacket {
|
||||
magic: CloudProtoMagic::TS,
|
||||
kind: TsPacketKind::Ack.into(),
|
||||
version: CloudProtoVersion::Normal,
|
||||
payload: txid.to_be_bytes().to_vec(),
|
||||
})?;
|
||||
let _ = this.unacked_txid.take();
|
||||
|
||||
// If the ACK doesn't finish leaving here, that's fine,
|
||||
// we also flush below when our io's recv side is still Pending
|
||||
ready!(this.io.poll_flush_unpin(cx))?;
|
||||
}
|
||||
if let Some(ev) = this.unacked_event.take() {
|
||||
assert!(this.unacked_txid.is_none());
|
||||
return Poll::Ready(Some(Ok(ev)));
|
||||
}
|
||||
|
||||
'_receive_packets: loop {
|
||||
let pkt = match this.io.poll_next_unpin(cx) {
|
||||
Poll::Ready(Some(pkt)) => pkt,
|
||||
Poll::Ready(None) => return Poll::Ready(None),
|
||||
Poll::Pending => {
|
||||
// If the user is only polling the read side, some of our ACKs might never finish flushing,
|
||||
// the other server would stop sending, and this poll_next would be Pending forever :)
|
||||
// So if we have nothing left but the user is still reading, it's a good time to flush our send side
|
||||
ready!(this.io.poll_flush_unpin(cx))?;
|
||||
return Poll::Pending; // We still have a queued wake on the read side
|
||||
}
|
||||
};
|
||||
|
||||
if pkt.kind == TsPacketKind::Ack {
|
||||
// This would be the place to update a queue of un-ACKed inflight packets,
|
||||
// so we can have backpressure, and retransmits packets after some time.
|
||||
//
|
||||
// We don't do any of that, because Crowdstrike's client doesn't either,
|
||||
// and it's unreasonably hard to be the only side "following TCP rules"
|
||||
// if the other side assumes packets it sends can never be dropped.
|
||||
//
|
||||
// See the other (large) comment below on the send side for more context.
|
||||
if pkt.payload.len() == 8 {
|
||||
let txid = u64::from_be_bytes(pkt.payload[..].try_into().unwrap());
|
||||
trace!("Received ACK for event txid {:#x}", txid);
|
||||
} else {
|
||||
error!(
|
||||
"Received ACK packet with invalid size: {:#x}",
|
||||
pkt.payload.len()
|
||||
)
|
||||
}
|
||||
continue;
|
||||
} else if pkt.kind == TsPacketKind::Event {
|
||||
if pkt.payload.len() < HDR_TXID_SIZE + EVT_HDR_LEN {
|
||||
return Poll::Ready(Some(Err(CloudProtoError::PayloadTooShort(
|
||||
pkt.payload.len(),
|
||||
HDR_TXID_SIZE + EVT_HDR_LEN,
|
||||
))));
|
||||
}
|
||||
let txid = u64::from_be_bytes(pkt.payload[..HDR_TXID_SIZE].try_into().unwrap());
|
||||
let ev = Event::from_read(&mut Cursor::new(&pkt.payload[HDR_TXID_SIZE..]))?;
|
||||
|
||||
// We ACK received events before returning them, to make sure we keep getting polled until the ACK is sent
|
||||
// So we have to buffer the event and its txid, in case we get Poll::Pending while trying to ACK it
|
||||
trace!(
|
||||
"Received event with txid {:#x}, preparing to send ACK",
|
||||
txid
|
||||
);
|
||||
assert!(this.unacked_txid.is_none());
|
||||
this.unacked_txid = Some(txid);
|
||||
assert!(this.unacked_event.is_none());
|
||||
this.unacked_event = Some(ev);
|
||||
continue 'process_pending_acks;
|
||||
} else {
|
||||
// Hoping this was a non-essential packet and continuing happily...
|
||||
warn!(
|
||||
"Received unexpected CloudProto packet kind: {:#x}",
|
||||
pkt.kind
|
||||
);
|
||||
trace!("Unexpected packet payload: {}", hex::encode(&pkt.payload));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<IO> Sink<Event> for TsEventSocket<IO>
|
||||
where
|
||||
IO: AsyncRead + AsyncWrite,
|
||||
{
|
||||
type Error = std::io::Error;
|
||||
|
||||
fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
// If we wanted to tracked ACKs for our tx, here we would need to block when the
|
||||
// queue of inflight un-ACKed events we're trackign becomes full.
|
||||
// But that queue can only shrink when we *receive* ACKs, so the TX side would depend
|
||||
// on Rust users of the lib also polling the RX side regularly, *while* they send.
|
||||
// If a user did some ts_sock.send().await in a loop without ever receiving, we'd DEADLOCK.
|
||||
//
|
||||
// So we could poll the RX internally from TX when our send queue is full until we get ACKs
|
||||
// But it's a single RX stream. We might receive real Events instead, all while inside send.
|
||||
// So we could stash them in an RX queue until full, but then we what should send() do?
|
||||
//
|
||||
// An unsatisfying option is to just return an error from start_send() when that happens,
|
||||
// because your RX queue is full, maybe no one's polling RX, and we don't want to deadlock.
|
||||
// But that error could also happen in normal usage if the TX races faster than the RX.
|
||||
// And in Rust, Sinks that return error are generally expected to be permanently closed.
|
||||
// Rust sinks don't normally do "oops you need to poll the read side a little!" errors :)
|
||||
//
|
||||
// This is where TX would just drop packets on the floor when the RX queue is full,
|
||||
// and that ought to be fine! If we drop them they won't be ACKd, and the other side
|
||||
// will just retransmit them to us later. A little bit of inefficiency on the RX side,
|
||||
// but this only happens when the RX queue is full, so that'd be reasonable.
|
||||
//
|
||||
// Except, as it turns out, the official Crowdstrike client does **none of that**!
|
||||
//
|
||||
// It *looks like* it has code to do it! It *seems* to track inflight packets,
|
||||
// walks all sorts of linked lists to process inflight packets and the ACKs for them,
|
||||
// its send worker thread has some kind of "send window" and "backpressure" like routines.
|
||||
//
|
||||
// But in practice, it completely ignores ACKs. It always just keeps forging ahead,
|
||||
// so we can't rely on being able to drop packets when our RX queue is full, they'd be lost.
|
||||
// This is the problem with being the only side trying to uphold guarantees, you only
|
||||
// get the constraints but you don't get to rely on the other side following them ^^'
|
||||
//
|
||||
// This is still work-around-able, either by telling users to always split() this socket
|
||||
// and always poll the RX side in an async task, so we get to just block in TX and be done.
|
||||
// This is pretty much how Crowdstrike's client is architectured too.
|
||||
// It has an RX and TX worker, and reality aside, in theory it should just work like this.
|
||||
//
|
||||
// Or, we could do "engineering" and add a 100ms timer since the last RX poll.
|
||||
// TX would only try to go receive ACKs itself after 100ms pass without any RX poll,
|
||||
// it'd put Events in the RX queue until full, and return an error when that's full.
|
||||
// Because of the timer this can't happen in normal usage anymore.
|
||||
// With this, if you are polling RX from time to time TX will just wait for you.
|
||||
// If you're only sending but you know you won't be received packets, it
|
||||
// TX *will* have to do reads, but it will see only ACKs, won't have to fill the RX queue,
|
||||
// and so you will not see the error in practice.
|
||||
// Only if you keep sending without having an RX worker, and the other side actually
|
||||
// replies with Events, then we'd return the error, because that's still better
|
||||
// than a deadlock or just dropping received packets that the other side won't retransmit.
|
||||
//
|
||||
//
|
||||
// But, instead of all that, we just do as the romans do and happily ignore received ACKs.
|
||||
// Firstly because it's completely unnecessary, the Crowdstrike server already has to
|
||||
// deal with a client that doesn't follow ACKs at all, so we're "bug-for-bug" compatible.
|
||||
//
|
||||
// Second because CLOUDPROTO is *always* carried over TLS, so ACKs were never necessary
|
||||
// in the first place! Even if you do TLS over UDP for some reason, you will already have
|
||||
// done retransmissions below the TLS layer, because TLS won't just let you drop packets in
|
||||
// the middle of an encrypted stream. The crypto layer tends to not like that idea.
|
||||
//
|
||||
// It's interesting that their client _almost_ implements all of this machinery,
|
||||
// but then gives up in practice. Maybe it was just too complex and/or annoying to debug?
|
||||
// I'm just curious why they still ship all of this code that doesn't run in practice...
|
||||
// For instance, the client has a check where if a function returns some status code,
|
||||
// that means a duplicate ACK was received. Except that status code is _never_ returned
|
||||
// by that function, in fact it's found nowhere else in the code accoring to IDA :)
|
||||
//
|
||||
// A lot of the client code is like this, half implemented stuff. But maybe we should
|
||||
// really be impressed by this surely purposeful obfuscation and misdirection.
|
||||
// (...almost as effective as having to follow all those damn C++ virtual calls everywhere!)
|
||||
let this = self.get_mut();
|
||||
this.io.poll_ready_unpin(cx)
|
||||
}
|
||||
|
||||
fn start_send(self: Pin<&mut Self>, ev: Event) -> Result<(), Self::Error> {
|
||||
let this = self.get_mut();
|
||||
|
||||
let mut buf = Vec::with_capacity(HDR_TXID_SIZE + EVT_HDR_LEN + ev.data.len());
|
||||
buf.extend_from_slice(&this.next_txid.to_be_bytes());
|
||||
this.next_txid += TXID_INCREMENT;
|
||||
match ev.into_write(&mut buf) {
|
||||
Ok(_) => {}
|
||||
Err(CloudProtoError::Io { source }) => return Err(source),
|
||||
Err(e) => {
|
||||
return Err(std::io::Error::new(
|
||||
std::io::ErrorKind::Other,
|
||||
format!("Unexpected error while sending Event: {}", e),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
this.io.start_send_unpin(CloudProtoPacket {
|
||||
magic: CloudProtoMagic::TS,
|
||||
kind: TsPacketKind::Event.into(),
|
||||
version: CloudProtoVersion::Normal,
|
||||
payload: buf,
|
||||
})
|
||||
}
|
||||
|
||||
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
self.get_mut().io.poll_flush_unpin(cx)
|
||||
}
|
||||
|
||||
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
self.get_mut().io.poll_close_unpin(cx)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user