Add CloudProtoSocket carrying CloudProtoPackets

.gitignore

@@ -1,2 +1,3 @@
 /target
 /Cargo.lock
+client_keys.log

Cargo.toml

@@ -3,6 +3,21 @@ name = "crowdstrike-cloudproto"
 version = "0.1.0"
 edition = "2021"
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
 [dependencies]
+tokio = { version = "1", features = ["io-util"] }
+tokio-util = { version = "0.7.3", features = ["codec"] }
+futures-util = { version = "0.3.23", features = ["sink"] }
+bytes = "1.2.1"
+byteorder = "1.4.3"
+thiserror = "1.0.32"
+tracing = "0.1.36"
+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 }
+tracing-subscriber = { version = "0.3.15", features = ["env-filter", "fmt"] }

src/framing.rs

@@ -0,0 +1,33 @@
+//! This module provides an async [`CloudProtoSocket`](socket::CloudProtoSocket) Stream + Sink that handles [`CloudProtoPacket`](packet::CloudProtoPacket)s.
+//!
+//! CLOUDPROTO is a packet-based big endian binary protocol that transports events or other payloads.
+//! The framing layer handles the common outer header/framing,
+//! but ignores the inner service-specific payload format and interpretation of packet kinds.
+
+mod hdr_version;
+pub mod packet;
+mod socket;
+
+pub use hdr_version::CloudProtoVersion;
+pub use socket::CloudProtoSocket;
+
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum CloudProtoError {
+    #[error("Bad CloudProto magic {0:#x}, expected {1:#x}")]
+    BadMagic(u8, u8),
+    #[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),
+    #[error("{0}")]
+    ClosedByPeer(String),
+    #[error("CloudProto IO error")]
+    Io {
+        #[from]
+        source: std::io::Error,
+    },
+}

src/framing/hdr_version.rs

@@ -0,0 +1,56 @@
+#[repr(u16)]
+#[derive(Eq, PartialEq, Copy, Clone, Debug)]
+#[cfg_attr(test, derive(strum_macros::EnumCount))]
+pub enum CloudProtoVersion {
+    /// All packets that don't fall in other categories
+    Normal,
+    /// Used for the first packets sent
+    Connect,
+    /// I haven't seen other values used yet
+    Other(u16),
+}
+
+impl From<u16> for CloudProtoVersion {
+    fn from(value: u16) -> Self {
+        match value {
+            x if x == Self::Normal.into() => Self::Normal,
+            x if x == Self::Connect.into() => Self::Connect,
+            x => Self::Other(x),
+        }
+    }
+}
+
+impl From<CloudProtoVersion> for u16 {
+    fn from(kind: CloudProtoVersion) -> Self {
+        match kind {
+            CloudProtoVersion::Normal => 1,
+            CloudProtoVersion::Connect => 2,
+            CloudProtoVersion::Other(x) => x,
+        }
+    }
+}
+
+impl From<&CloudProtoVersion> for u16 {
+    fn from(v: &CloudProtoVersion) -> Self {
+        u16::from(*v)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use crate::framing::CloudProtoVersion;
+    use std::collections::HashSet;
+    use strum::EnumCount;
+
+    #[test]
+    fn cloud_proto_magic_roundtrip() {
+        let mut seen = HashSet::new();
+        for v in 0..=u16::MAX {
+            let m = CloudProtoVersion::from(v);
+            seen.insert(std::mem::discriminant(&m));
+            assert_eq!(u16::from(m), v);
+        }
+        // If this fails, you may have forgotten to update From<u16>
+        assert_eq!(seen.len(), CloudProtoVersion::COUNT)
+    }
+}

src/framing/packet.rs

@@ -0,0 +1,79 @@
+use crate::framing::{CloudProtoError, CloudProtoVersion};
+use crate::services::CloudProtoMagic;
+use byteorder::{ReadBytesExt, BE};
+use std::io::Cursor;
+
+pub(crate) const COMMON_HDR_LEN: usize = 8;
+
+/// The common framing packet structure of the protocol
+#[derive(Eq, PartialEq, Debug, Clone)]
+pub struct CloudProtoPacket {
+    /// One magic value corresponds to one backend service
+    pub magic: CloudProtoMagic,
+    /// 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>,
+}
+
+impl CloudProtoPacket {
+    pub(crate) fn from_buf(buf: &[u8]) -> Result<Self, CloudProtoError> {
+        let mut reader = Cursor::new(buf);
+        let magic = reader.read_u8()?.into();
+        let kind = reader.read_u8()?;
+        let version = reader.read_u16::<BE>()?.into();
+        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));
+        }
+        let payload = buf[reader.position() as usize..].to_vec();
+        Ok(Self {
+            magic,
+            kind,
+            version,
+            payload,
+        })
+    }
+
+    pub(crate) fn to_buf(&self) -> Vec<u8> {
+        use byteorder::WriteBytesExt;
+        use std::io::Write;
+
+        let mut buf = Vec::new();
+        let mut writer = Cursor::new(&mut buf);
+        writer.write_u8(self.magic.into()).unwrap();
+        writer.write_u8(self.kind).unwrap();
+        writer.write_u16::<BE>(self.version.into()).unwrap();
+        writer
+            .write_u32::<BE>((self.payload.len() + COMMON_HDR_LEN) as u32)
+            .unwrap();
+        writer.write_all(&self.payload).unwrap();
+        writer.flush().unwrap();
+        buf
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use crate::framing::packet::CloudProtoPacket;
+    use crate::framing::CloudProtoVersion;
+    use crate::services::CloudProtoMagic;
+    use anyhow::Result;
+
+    #[test_log::test]
+    fn to_from_buf_serialization() -> Result<()> {
+        let pkt = CloudProtoPacket {
+            magic: CloudProtoMagic::Other(0xFF),
+            kind: 0x73,
+            version: CloudProtoVersion::Other(0x10E9),
+            payload: b"Hello world".to_vec(),
+        };
+        let pkt2 = CloudProtoPacket::from_buf(&pkt.to_buf())?;
+        assert_eq!(pkt, pkt2);
+
+        Ok(())
+    }
+}

src/framing/socket.rs

@@ -0,0 +1,133 @@
+use crate::framing::packet::CloudProtoPacket;
+use crate::framing::CloudProtoError;
+use bytes::Bytes;
+use futures_util::{Sink, SinkExt, Stream, StreamExt};
+use std::pin::Pin;
+use std::task::{ready, Context, Poll};
+use tokio::io::{AsyncRead, AsyncWrite, ReadHalf, WriteHalf};
+use tokio_util::codec::{BytesCodec, FramedRead, FramedWrite, LengthDelimitedCodec};
+use tracing::{error, trace};
+
+pub struct CloudProtoSocket<IO> {
+    read: FramedRead<ReadHalf<IO>, LengthDelimitedCodec>,
+    write: FramedWrite<WriteHalf<IO>, BytesCodec>,
+}
+
+impl<IO> CloudProtoSocket<IO>
+where
+    IO: AsyncRead + AsyncWrite + Unpin,
+{
+    pub async fn new(io: IO) -> Result<Self, CloudProtoError> {
+        let (read, write) = tokio::io::split(io);
+        let read = LengthDelimitedCodec::builder()
+            .big_endian()
+            .length_field_type::<u32>()
+            .length_adjustment(0)
+            .length_field_offset(4)
+            .num_skip(0)
+            .new_read(read);
+        let write = FramedWrite::new(write, BytesCodec::new());
+        Ok(Self { read, write })
+    }
+}
+
+impl<IO> Stream for CloudProtoSocket<IO>
+where
+    IO: AsyncRead + AsyncWrite + Unpin,
+{
+    type Item = CloudProtoPacket;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.get_mut();
+        loop {
+            let pkt = match ready!(this.read.poll_next_unpin(cx)) {
+                Some(Ok(frame)) => CloudProtoPacket::from_buf(&frame),
+                Some(Err(e)) => {
+                    error!("Failed to read cloudproto frame: {}", e);
+                    return Poll::Ready(None);
+                }
+                None => return Poll::Ready(None),
+            };
+            match pkt {
+                Ok(pkt) => {
+                    trace!(
+                        "Received kind 0x{:x} packet with 0x{:x} bytes payload: {}",
+                        pkt.kind,
+                        pkt.payload.len(),
+                        hex::encode(&pkt.payload),
+                    );
+                    return Poll::Ready(Some(pkt));
+                }
+                Err(e) => {
+                    error!("Bad cloudproto packet: {}", e);
+                    continue;
+                }
+            }
+        }
+    }
+}
+
+impl<IO> Sink<CloudProtoPacket> for CloudProtoSocket<IO>
+where
+    IO: AsyncRead + AsyncWrite + Unpin,
+{
+    type Error = std::io::Error;
+
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        SinkExt::<Bytes>::poll_ready_unpin(&mut self.get_mut().write, cx)
+    }
+
+    fn start_send(self: Pin<&mut Self>, pkt: CloudProtoPacket) -> Result<(), Self::Error> {
+        let this = self.get_mut();
+        let buf = Bytes::from(pkt.to_buf());
+        trace!(
+            "Sending kind 0x{:x} packet with 0x{:x} bytes payload: {}",
+            pkt.kind,
+            pkt.payload.len(),
+            hex::encode(&pkt.payload),
+        );
+        this.write.start_send_unpin(buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        SinkExt::<Bytes>::poll_flush_unpin(&mut self.get_mut().write, cx)
+    }
+
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        SinkExt::<Bytes>::poll_close_unpin(&mut self.get_mut().write, cx)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use crate::framing::packet::CloudProtoPacket;
+    use crate::framing::CloudProtoVersion;
+    use crate::services::CloudProtoMagic;
+    use crate::CloudProtoSocket;
+    use anyhow::Result;
+    use futures_util::{SinkExt, StreamExt};
+    use rand::Rng;
+
+    #[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 rng = rand::thread_rng();
+        let len = rng.gen::<u16>() as usize;
+        let mut payload = Vec::with_capacity(len);
+        payload.resize(len, len as u8);
+        let pkt = CloudProtoPacket {
+            magic: CloudProtoMagic::TS,
+            kind: 0,
+            version: CloudProtoVersion::Normal,
+            payload,
+        };
+        client.send(pkt.clone()).await?;
+        let reply = server.next().await.unwrap();
+        assert_eq!(pkt, reply);
+
+        Ok(())
+    }
+}

src/lib.rs

@@ -1,14 +1,46 @@
-pub fn add(left: usize, right: usize) -> usize {
-    left + right
-}
+//! 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.
 
-#[cfg(test)]
-mod tests {
-    use super::*;
+pub mod framing;
+pub mod services;
 
-    #[test]
-    fn it_works() {
-        let result = add(2, 2);
-        assert_eq!(result, 4);
-    }
-}
+pub use framing::CloudProtoSocket;

src/services.rs

@@ -0,0 +1,81 @@
+mod lfo;
+mod ts;
+
+pub use lfo::*;
+pub use ts::*;
+
+/// 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.
+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";
+/// Arbitrary machine-specific value generated on an isolated VM
+pub const DEFAULT_UNK0_HEX: &str = "54645dacc392cb43b4803094141e0087";
+
+#[repr(u8)]
+#[derive(Eq, PartialEq, Debug, Copy, Clone)]
+#[cfg_attr(test, derive(strum_macros::EnumCount))]
+pub enum CloudProtoMagic {
+    TS,
+    LFO,
+    Other(u8),
+}
+
+impl From<u8> for CloudProtoMagic {
+    fn from(value: u8) -> Self {
+        match value {
+            x if x == Self::TS => Self::TS,
+            x if x == Self::LFO => Self::LFO,
+            x => Self::Other(x),
+        }
+    }
+}
+
+impl From<CloudProtoMagic> for u8 {
+    fn from(kind: CloudProtoMagic) -> Self {
+        match kind {
+            CloudProtoMagic::TS => 0x8F,
+            CloudProtoMagic::LFO => 0x9F,
+            CloudProtoMagic::Other(x) => x,
+        }
+    }
+}
+
+impl From<&CloudProtoMagic> for u8 {
+    fn from(v: &CloudProtoMagic) -> Self {
+        u8::from(*v)
+    }
+}
+
+impl PartialEq<u8> for CloudProtoMagic {
+    fn eq(&self, other: &u8) -> bool {
+        u8::from(self) == *other
+    }
+}
+
+impl PartialEq<CloudProtoMagic> for u8 {
+    fn eq(&self, other: &CloudProtoMagic) -> bool {
+        u8::from(other) == *self
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::CloudProtoMagic;
+    use std::collections::HashSet;
+    use strum::EnumCount;
+
+    #[test]
+    fn cloud_proto_magic_roundtrip() {
+        let mut seen = HashSet::new();
+        for v in 0..=u8::MAX {
+            let m = CloudProtoMagic::from(v);
+            seen.insert(std::mem::discriminant(&m));
+            assert_eq!(u8::from(m), v);
+        }
+        // If this fails, you may have forgotten to update From<u8>
+        assert_eq!(seen.len(), CloudProtoMagic::COUNT)
+    }
+}

src/services/lfo.rs

@@ -0,0 +1,44 @@
+use crate::services::DEFAULT_CID_HEX;
+
+mod pkt_kind;
+
+#[derive(Eq, PartialEq, Debug, Clone)]
+pub struct LfoRequest {
+    // The CID assigned to a Crowdstrike customer (same as the CCID without the last -N number)
+    // The LFO server doesn't really check if it belongs to anyone. Just try to pass a valid CID.
+    pub(crate) cid: [u8; 16],
+    // Agent ID. LFO isn't uptight like TS if the AID is not an active customer.
+    // In fact, you can give it all zeroes. LFO is friendly like that.
+    pub(crate) aid: [u8; 16],
+    // The real client supports values 0 or 1. We only support 0.
+    pub(crate) compression: u16,
+    // The file to download
+    pub(crate) remote_path: String,
+    // This field is probably the offset for chunked downloads. Not supported or tested yet.
+    // Large files can't be downloaded in one packet, so the client may get partial responses
+    // The offset allows downloading the rest of those large files in multiple queries
+    pub(crate) offset: u32,
+}
+
+impl LfoRequest {
+    /// Create a request for `remote_path` with default values
+    pub fn new_simple(remote_path: String) -> Self {
+        Self {
+            cid: hex::decode(DEFAULT_CID_HEX).unwrap().try_into().unwrap(),
+            aid: [0; 16], // LFO doesn't mind all zeroes
+            compression: 0,
+            remote_path,
+            offset: 0,
+        }
+    }
+
+    pub fn new_custom(cid: [u8; 16], aid: [u8; 16], remote_path: String) -> Self {
+        Self {
+            cid,
+            aid,
+            compression: 0, // Only 0 supported for now
+            remote_path,
+            offset: 0, // Only 0 supported for now
+        }
+    }
+}

src/services/lfo/pkt_kind.rs

@@ -0,0 +1,89 @@
+/// 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))]
+pub enum LfoPacketKind {
+    /// Sent by client to request file data
+    GetFileRequest,
+    /// Successful response to a request
+    ReplyOk,
+    /// May contain a friendly error message (at offset 0x8).
+    /// If you send bad requests, you may get a ReplyFail with "internal error" (consider not doing that!)
+    /// If the request is sufficiently bad, the server may also just close the socket without replying
+    ReplyFail,
+    /// Other values have not been observed yet
+    Other(u8),
+}
+
+impl From<LfoPacketKind> for u8 {
+    fn from(kind: LfoPacketKind) -> Self {
+        match kind {
+            LfoPacketKind::GetFileRequest => 1,
+            LfoPacketKind::ReplyOk => 2,
+            LfoPacketKind::ReplyFail => 3,
+            LfoPacketKind::Other(x) => x,
+        }
+    }
+}
+
+impl From<&LfoPacketKind> for u8 {
+    fn from(pkt: &LfoPacketKind) -> Self {
+        u8::from(*pkt)
+    }
+}
+
+impl From<u8> for LfoPacketKind {
+    fn from(value: u8) -> Self {
+        match value {
+            x if x == Self::GetFileRequest => Self::GetFileRequest,
+            x if x == Self::ReplyOk => Self::ReplyOk,
+            x if x == Self::ReplyFail => Self::ReplyFail,
+            x => Self::Other(x),
+        }
+    }
+}
+
+impl PartialEq<u8> for LfoPacketKind {
+    fn eq(&self, other: &u8) -> bool {
+        u8::from(self) == *other
+    }
+}
+
+impl PartialEq<LfoPacketKind> for u8 {
+    fn eq(&self, other: &LfoPacketKind) -> bool {
+        u8::from(other) == *self
+    }
+}
+
+impl std::fmt::LowerHex for LfoPacketKind {
+    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 LfoPacketKind {
+    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::LfoPacketKind;
+    use std::collections::HashSet;
+    use strum::EnumCount;
+
+    #[test]
+    fn lfo_kind_roundtrip() {
+        let mut seen = HashSet::new();
+        for v in 0..=u8::MAX {
+            let m = LfoPacketKind::from(v);
+            seen.insert(std::mem::discriminant(&m));
+            assert_eq!(u8::from(m), v);
+        }
+        // If this fails, you may have forgotten to update From<u8>
+        assert_eq!(seen.len(), LfoPacketKind::COUNT)
+    }
+}

src/services/ts.rs

@@ -0,0 +1,53 @@
+mod pkt_kind;
+pub use pkt_kind::TsPacketKind;
+
+use crate::services::{DEFAULT_BOOTID_HEX, DEFAULT_UNK0_HEX};
+
+/// Connection information required to open a session with the TS server
+#[derive(Eq, PartialEq, Debug, Clone)]
+pub struct TsConnectInfo {
+    // The CID assigned to a Crowdstrike customer (same as the CCID without the last -N number)
+    // These are not random, there's a sort of checksum that must pass for a CID to be valid.
+    // For TS the CID needs to be not only valid, but belong to an active customer
+    pub(crate) cid: [u8; 16],
+    // Unknown, but has never changed and the AID returned by TS depends on it (can also be 0)
+    pub(crate) unk0: [u8; 16],
+    // Agent ID. Saved in "falconstore". New values can be assigned by the TS server on connection
+    pub(crate) aid: [u8; 16],
+    // Per-machine value (the stable /proc/sys/kernel/random/boot_id, or a timestamp if unavailable)
+    pub(crate) bootid: [u8; 16],
+    // The "PT" value from "falconstore". Can be left as zeroes.
+    pub(crate) pt: [u8; 8],
+}
+
+impl TsConnectInfo {
+    /// Connect using the provided Crowdstrike customer ID
+    /// The CID must belong to an active customer.
+    /// Unlike for the LSO server and falcon-sensor it's not enough to use a structurally valid but inactive CID.
+    /// Uses hardcoded default values for the other non-critical fields.
+    pub fn new_simple(cid: [u8; 16]) -> Self {
+        Self {
+            cid,
+            unk0: hex::decode(DEFAULT_UNK0_HEX).unwrap().try_into().unwrap(),
+            aid: [0; 16],
+            bootid: hex::decode(DEFAULT_BOOTID_HEX).unwrap().try_into().unwrap(),
+            pt: [0; 8],
+        }
+    }
+
+    pub fn new_custom(
+        cid: [u8; 16],
+        unk0: [u8; 16],
+        aid: [u8; 16],
+        bootid: [u8; 16],
+        pt: [u8; 8],
+    ) -> Self {
+        Self {
+            cid,
+            unk0,
+            aid,
+            bootid,
+            pt,
+        }
+    }
+}

src/services/ts/pkt_kind.rs

@@ -0,0 +1,94 @@
+/// 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))]
+pub enum TsPacketKind {
+    /// First packet from client to server
+    Connect,
+    /// First reply from server to client
+    ConnectionEstablished,
+    /// 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.
+    Ack,
+    /// This escape hatch is provided with no warranty including fitness for a particular purpose.
+    /// Good luck!
+    Other(u8),
+}
+
+impl From<TsPacketKind> for u8 {
+    fn from(kind: TsPacketKind) -> Self {
+        match kind {
+            TsPacketKind::Connect => 1,
+            TsPacketKind::ConnectionEstablished => 2,
+            TsPacketKind::Event => 3,
+            TsPacketKind::Ack => 4,
+            TsPacketKind::Other(x) => x,
+        }
+    }
+}
+
+impl From<&TsPacketKind> for u8 {
+    fn from(pkt: &TsPacketKind) -> Self {
+        u8::from(*pkt)
+    }
+}
+
+impl From<u8> for TsPacketKind {
+    fn from(value: u8) -> Self {
+        match value {
+            x if x == Self::Connect => Self::Connect,
+            x if x == Self::ConnectionEstablished => Self::ConnectionEstablished,
+            x if x == Self::Event => Self::Event,
+            x if x == Self::Ack => Self::Ack,
+            x => Self::Other(x),
+        }
+    }
+}
+
+impl PartialEq<u8> for TsPacketKind {
+    fn eq(&self, other: &u8) -> bool {
+        u8::from(self) == *other
+    }
+}
+
+impl PartialEq<TsPacketKind> for u8 {
+    fn eq(&self, other: &TsPacketKind) -> bool {
+        u8::from(other) == *self
+    }
+}
+
+impl std::fmt::LowerHex for TsPacketKind {
+    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 TsPacketKind {
+    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::TsPacketKind;
+    use std::collections::HashSet;
+    use strum::EnumCount;
+
+    #[test]
+    fn ts_kind_roundtrip() {
+        let mut seen = HashSet::new();
+        for v in 0..=u8::MAX {
+            let m = TsPacketKind::from(v);
+            seen.insert(std::mem::discriminant(&m));
+            assert_eq!(u8::from(m), v);
+        }
+        // If this fails, you may have forgotten to update From<u8>
+        assert_eq!(seen.len(), TsPacketKind::COUNT)
+    }
+}