Feeds

Stream each feed's real-time events and download its hourly and daily Parquet exports.

Each feed is a Synthient::Feed returned by a reader on Client. Every feed has a real-time stream and hourly Parquet exports that roll up into daily exports. Streaming and exports need separate scopes on the API key. In the RBS signatures, Feed is generic over its event type, so client.proxies is a Feed[ProxyEvent]. The event types are listed under Event types.

What each feed carries:

• proxies carries proxy sightings.
• anonymizers carries VPN and anonymizer address ranges.
• torrents carries torrents seen on the BitTorrent DHT and trackers.
• ja4t carries JA4T TCP fingerprint sightings.
• The four Helios feeds carry honeypot captures: HTTP requests, TLS ClientHellos, the destinations tunnels were asked to reach, and Android Debug Bridge shell commands. Despite its name, helios_dns does not carry DNS queries.

See Firehose, Feeds, and Helios for the data behind each feed.

Streams

stream

Yields the feed's real-time events as they happen, each decoded into the feed's event type E.

def stream: () { (E event) -> void } -> nil
          | () -> Enumerator[E, nil]

• Connection: Each call opens a new connection. The server keeps it open for up to 30 minutes and then closes it, and stream returns nil. Breaking out of the block closes the connection.
• Errors: Exceptions raised in the block pass through unchanged. A response other than 200 raises an APIError, a dropped connection raises a ConnectionError, and an event that does not match the schema raises a DecodeError. A 429 APIError without retry_after means the key's monthly request budget is spent or it already has too many open streams.
• Reconnecting: The API has no resume cursor, so events sent while disconnected are missed.

Call stream again in a loop to reconnect:

loop do
  client.proxies.stream do |event|
    puts "#{event.ip} #{event.provider} #{event.type}"
  end
end

The loop reconnects immediately after a clean close. Opening a connection already uses the SDK's bounded retries, including Retry-After handling. Any error that remains after those retries, or occurs while reading the stream, stops this consumer. A 429 without Retry-After can indicate an exhausted monthly request budget or too many open streams; resolve that condition before reconnecting.

Without a block, stream returns an Enumerator that opens the connection when iterated, so Enumerable methods work on it:

sample = client.proxies.stream.first(100)
puts sample.map(&:provider).tally

Every feed streams the same way. Only the event type changes:

client.proxies.stream do |event|
  puts "#{event.ip} #{event.provider} #{event.type} #{event.country_code} #{event.asn}"
end

HeliosHTTPSEvent#details is nil when the sensor could not parse the handshake, and HeliosHTTPEvent#details is optional too, so check them before use.

Exports

Each feed has hourly Parquet exports that roll up into daily exports.

• Hourly exports: An hourly export is available until its day's daily export is rolled up, shortly after 00:30 UTC the next day. After that, requests for the hour raise a 400 APIError.
• Daily exports: While a daily export is being rolled up, requests for it raise a 404 APIError with retry_after set.

Export IDs

An export ID is one of:

• "latest"
• a UTC day as a Date or "YYYY-MM-DD"
• a UTC hour as a Time or "YYYY-MM-DD/H", where a zero-padded hour is also accepted

type export_id = String | Date | Time

• Latest: "latest" is the newest hourly export during the day, or the daily export once that day's hours are rolled up.
• Input: A Time is converted to UTC and identifies the hour that contains it. Export#id values can be passed as they are. Any other string raises ArgumentError, and a DateTime or any other type raises TypeError, before a request is sent.
• Helpers: The SDK has no ID helper methods or constant for "latest": pass a Date, a Time, or a string.

yesterday = Time.now.utc.to_date - 1
last_hour = Time.now - 3600
puts client.proxies.export_meta(yesterday).rows
puts client.proxies.export_meta(last_hour).rows

list_exports

Returns one page of the feed's exports, newest first.

def list_exports: (?limit: Integer?, ?cursor: String?) -> ExportPage

page = client.proxies.list_exports(limit: 24)
page.exports.each do |export|
  puts "#{export.id} #{export.kind} #{export.row_count}"
end
puts "next cursor: #{page.next_cursor}"

• Paging: Use each_export to iterate over every page.

each_export

Yields every export, newest first, fetching pages as they are needed.

def each_export: (?limit: Integer?, ?cursor: String?) { (Export export) -> void } -> nil
               | (?limit: Integer?, ?cursor: String?) -> Enumerator[Export, nil]

client.proxies.each_export do |export|
  puts "#{export.id} #{export.kind} #{export.size_bytes}"
end

daily = client.proxies.each_export.find { |export| export.kind == Synthient::ExportKind::DAILY }
puts daily&.id

• Paging: Breaking out of the block stops fetching.
• Enumerator: Without a block, it returns an Enumerator.
• Input: limit and cursor work as they do for list_exports.
• Errors: If the API returns a cursor it already returned, each_export raises Synthient::Error.

export_meta

Returns the checksum, size, row count, and Parquet schema of an export.

def export_meta: (export_id export) -> ExportMeta

meta = client.proxies.export_meta("latest")
puts "#{meta.id} #{meta.rows} #{meta.size} #{meta.checksum}"
meta.schema.fields.each do |column|
  puts "#{column.name} #{column.type}"
end

download_url

Returns a presigned URL for an export's Parquet file.

def download_url: (export_id export) -> URI::HTTP

url = client.proxies.download_url("latest")
puts url.host

• Lifetime: The URL is valid for 24 hours.
• Security: The URL grants access without an API key, so treat it as a secret.
• Use: It is useful for tools that read Parquet over HTTPS directly.
• Errors: An error status raises an APIError. A success that is not a redirect, a redirect without a valid URL, or a redirect from an https base URL to plain http raises DecodeError.

download

Downloads an export, yielding the Parquet file's bytes in binary chunks as they arrive.

def download: (export_id export) { (String chunk) -> void } -> nil
            | (export_id export) -> Enumerator[String, nil]

• Connection: Breaking out of the block closes the connection.
• Enumerator: Without a block, it returns an Enumerator.
• Security: The API key is sent only to the Synthient API, never to the storage host that serves the file.
• Redirects: Up to 10 storage redirects are followed, never from https to http.
• Errors: A redirect the SDK cannot follow raises Synthient::Error. A storage response other than 200 or a redirect raises an APIError. Downloads from the storage host are not retried.

The SDK does not verify the file. Compare its SHA-256 with Export#checksum or ExportMeta#checksum:

require "digest"

yesterday = Time.now.utc.to_date - 1
meta = client.proxies.export_meta(yesterday)
digest = Digest::SHA256.new
File.open("proxies.parquet", "wb") do |file|
  client.proxies.download(yesterday) do |chunk|
    file.write(chunk)
    digest << chunk
  end
end
raise "checksum mismatch" unless digest.hexdigest == meta.checksum