Client Libraries - certstream-server-rust

Code Examples

Quick start examples in popular languages

                            Python
                            pip install certstream
                        

import certstream

def callback(message, context):
    if message['message_type'] == 'certificate_update':
        domains = message['data']['leaf_cert']['all_domains']
        print(domains)

certstream.listen_for_events(
    callback,
    url='wss://ws.certstream.dev/'
)

                            JavaScript
                            npm install ws
                        

const WebSocket = require('ws');
const ws = new WebSocket('wss://ws.certstream.dev/');

ws.on('message', (data) => {
    const msg = JSON.parse(data);
    if (msg.message_type === 'certificate_update') {
        console.log(msg.data.leaf_cert.all_domains);
    }
});

                            Go
                            certstream-go
                        

import "github.com/CaliDog/certstream-go"

func main() {
    stream, _ := certstream.CertStreamEventStream(false)

    for event := range stream {
        fmt.Println(event.Data.LeafCert.AllDomains)
    }
}

Available Clients

Official and community libraries

Python

Official certstream client

pip install certstream

JavaScript / Node.js

Any WebSocket library

npm install ws

Go

CaliDog certstream-go

go get github.com/CaliDog/certstream-go

CLI

Quick testing with wscat

wscat -c wss://ws.certstream.dev/

Browser

Native WebSocket API

new WebSocket('wss://ws.certstream.dev/')

cURL

SSE endpoint for scripts

curl -N https://ws.certstream.dev/sse

Advanced Examples

Filtering and processing certificates

                            Python
                            Phishing Detection
                        

import certstream
import re

BRAND_PATTERNS = [
    r'paypal', r'google', r'microsoft',
    r'apple', r'amazon', r'netflix'
]

def detect_phishing(message, context):
    if message['message_type'] != 'certificate_update':
        return

    domains = message['data']['leaf_cert']['all_domains']

    for domain in domains:
        for pattern in BRAND_PATTERNS:
            if re.search(pattern, domain, re.I):
                print(f'[ALERT] {domain}')

certstream.listen_for_events(
    detect_phishing,
    url='wss://ws.certstream.dev/'
)

                            JavaScript
                            Domain Monitoring
                        

const WebSocket = require('ws');

const WATCH_DOMAINS = ['mycompany.com', 'mybrand'];
const ws = new WebSocket('wss://ws.certstream.dev/');

ws.on('message', (data) => {
    const msg = JSON.parse(data);
    if (msg.message_type !== 'certificate_update') return;

    const domains = msg.data.leaf_cert.all_domains;

    for (const domain of domains) {
        for (const watch of WATCH_DOMAINS) {
            if (domain.includes(watch)) {
                console.log(`[MATCH] ${domain}`);
            }
        }
    }
});