← back to Domain Sniper
ct-source-ctlog: multi-log parallel polling via startMulti()
008b1caf3c66f5341d7b07219cf56c2c5a7be5b7 · 2026-05-12 16:31:36 -0700 · Steve Abrams
CT_LOGS=argon,xenon,wyvern env (default) → spawn 3 polling loops against a
shared EventEmitter, staggered by pollMs/N to avoid bursting all logs at the
same instant. Per-log cursors in certspotter-brand-cursors.json (already keyed
by log name). Consumers (brand-typo-watcher, dashboard) auto-detect CT_LOGS
vs CT_LOG and route to start()/startMulti().
Smoke (15s, argon+xenon+wyvern): 61 events ~= 4/sec firehose. vs ~2.2/sec
single-log = 1.8x. brand-typo PID 35012, dashboard PID 35013 restarted on
3-log multi-source.
Also: 2.9h honeypot spot-check 0/10 sniped — snipe-timing curve still flat.
Files touched
M brand-typo-watcher.jsM ct-source-ctlog.jsM dashboard.js
Diff
commit 008b1caf3c66f5341d7b07219cf56c2c5a7be5b7
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Tue May 12 16:31:36 2026 -0700
ct-source-ctlog: multi-log parallel polling via startMulti()
CT_LOGS=argon,xenon,wyvern env (default) → spawn 3 polling loops against a
shared EventEmitter, staggered by pollMs/N to avoid bursting all logs at the
same instant. Per-log cursors in certspotter-brand-cursors.json (already keyed
by log name). Consumers (brand-typo-watcher, dashboard) auto-detect CT_LOGS
vs CT_LOG and route to start()/startMulti().
Smoke (15s, argon+xenon+wyvern): 61 events ~= 4/sec firehose. vs ~2.2/sec
single-log = 1.8x. brand-typo PID 35012, dashboard PID 35013 restarted on
3-log multi-source.
Also: 2.9h honeypot spot-check 0/10 sniped — snipe-timing curve still flat.
---
brand-typo-watcher.js | 15 ++++++++------
ct-source-ctlog.js | 55 ++++++++++++++++++++++++++++++++++++++++-----------
dashboard.js | 14 +++++++------
3 files changed, 61 insertions(+), 23 deletions(-)
diff --git a/brand-typo-watcher.js b/brand-typo-watcher.js
index f37587c..e174755 100644
--- a/brand-typo-watcher.js
+++ b/brand-typo-watcher.js
@@ -22,6 +22,8 @@ const CERTSTREAM_URL = process.env.CERTSTREAM_URL || 'wss://certstream.calidog.i
// CT_SOURCE=certstream → original WebSocket flow (kept as fallback)
const CT_SOURCE = process.env.CT_SOURCE || 'ctlog';
const CT_LOG = process.env.CT_LOG || 'argon';
+// CT_LOGS comma-separated → multi-log parallel polling (default: argon+xenon+wyvern, ~4 cert events/sec)
+const CT_LOGS = (process.env.CT_LOGS || 'argon,xenon,wyvern').split(',').map((s) => s.trim()).filter(Boolean);
const CTLOG_POLL_MS = parseInt(process.env.CTLOG_POLL_MS || '5000', 10);
const CTLOG_BATCH = parseInt(process.env.CTLOG_BATCH || '32', 10);
const DATA_DIR = path.join(__dirname, 'data');
@@ -169,17 +171,18 @@ function connectCertstream() {
}
function connectCtLog() {
- console.error(`[${ts()}] CT_SOURCE=ctlog — polling log=${CT_LOG} poll=${CTLOG_POLL_MS}ms batch=${CTLOG_BATCH}, watching ${BRAND_KEYWORDS.length} brands (max-distance ${MAX_DISTANCE}, notify=${NOTIFY})`);
+ const opts = CT_LOGS.length > 1
+ ? { logKeys: CT_LOGS, pollMs: CTLOG_POLL_MS, batchSize: CTLOG_BATCH }
+ : { logKey: CT_LOG, pollMs: CTLOG_POLL_MS, batchSize: CTLOG_BATCH };
+ console.error(`[${ts()}] CT_SOURCE=ctlog — ${CT_LOGS.length > 1 ? 'multi=[' + CT_LOGS.join(',') + ']' : 'log=' + CT_LOG} poll=${CTLOG_POLL_MS}ms batch=${CTLOG_BATCH}, watching ${BRAND_KEYWORDS.length} brands (max-distance ${MAX_DISTANCE}, notify=${NOTIFY})`);
ctLogSource.start({
- logKey: CT_LOG,
- pollMs: CTLOG_POLL_MS,
- batchSize: CTLOG_BATCH,
+ ...opts,
onEvent: handleCertEvent,
onStatus: (st) => {
if (st.type === 'sth_err' || st.type === 'fetch_err' || st.type === 'error') {
- console.error(`[${ts()}] ctlog ${st.type}: ${st.msg || JSON.stringify(st)}`);
+ console.error(`[${ts()}] ctlog ${st.type} on ${st.log || '?'}: ${st.msg || JSON.stringify(st)}`);
} else if (st.type === 'seeded') {
- console.error(`[${ts()}] seeded at tree_size=${st.position}`);
+ console.error(`[${ts()}] seeded ${st.log} at tree_size=${st.position}`);
}
},
});
diff --git a/ct-source-ctlog.js b/ct-source-ctlog.js
index 2c3f518..f522785 100644
--- a/ct-source-ctlog.js
+++ b/ct-source-ctlog.js
@@ -219,14 +219,41 @@ function extractDnsNamesFromTBS(tbs) {
}
}
-function start({
+// Multi-log fan-out: if logKeys (array) is given, spawn one polling loop per
+// log against the SAME EventEmitter — consumers receive a merged firehose.
+// Returns a wrapper {stop, stats} that aggregates across logs.
+function startMulti({ logKeys, pollMs = 5_000, batchSize = 32, onEvent, onStatus } = {}) {
+ const ee = new EventEmitter();
+ if (onEvent) ee.on('certificate_update', onEvent);
+ if (onStatus) ee.on('status', onStatus);
+ const subs = logKeys.map((k, i) => startSingle({
+ logKey: k, pollMs, batchSize,
+ // Stagger polls so we don't burst all logs at the same instant
+ initialDelayMs: i * Math.max(500, Math.floor(pollMs / logKeys.length)),
+ sink: ee,
+ }));
+ return {
+ on: (...a) => ee.on(...a),
+ stop: () => subs.forEach((s) => s.stop()),
+ stats: () => subs.map((s) => ({ log: s.logName, stats: s.stats() })),
+ };
+}
+
+function start(opts = {}) {
+ if (Array.isArray(opts.logKeys) && opts.logKeys.length > 1) return startMulti(opts);
+ return startSingle(opts);
+}
+
+function startSingle({
logKey = 'argon',
pollMs = 5_000,
batchSize = 32,
+ initialDelayMs = 0,
+ sink, // optional shared EventEmitter for multi-log fan-out
onEvent,
onStatus,
} = {}) {
- const ee = new EventEmitter();
+ const ee = sink || new EventEmitter();
if (onEvent) ee.on('certificate_update', onEvent);
if (onStatus) ee.on('status', onStatus);
@@ -315,32 +342,38 @@ function start({
setTimeout(tick, wait);
}
}
- tick();
+ if (initialDelayMs > 0) setTimeout(tick, initialDelayMs); else tick();
return {
on: (...a) => ee.on(...a),
- stop: () => { stopped = true; ee.emit('status', { type: 'stopped', stats }); },
+ stop: () => { stopped = true; ee.emit('status', { type: 'stopped', log: log.name, stats }); },
stats: () => ({ ...stats }),
position: () => position,
+ logName: log.name,
};
}
-module.exports = { start, LOGS, parseLeaf, extractDnsNames, extractDnsNamesFromTBS };
+module.exports = { start, startMulti, LOGS, parseLeaf, extractDnsNames, extractDnsNamesFromTBS };
if (require.main === module) {
- const logKey = process.env.CT_LOG || 'argon';
const pollMs = parseInt(process.env.POLL_MS || '5000', 10);
const batchSize = parseInt(process.env.BATCH || '32', 10);
- console.log(`[${new Date().toISOString()}] ct-source-ctlog starting log=${logKey} poll=${pollMs}ms batch=${batchSize}`);
+ const ctLogs = process.env.CT_LOGS; // comma-separated for multi-log
+ const ctLog = process.env.CT_LOG || 'argon';
+ const opts = ctLogs
+ ? { logKeys: ctLogs.split(',').map((s) => s.trim()), pollMs, batchSize }
+ : { logKey: ctLog, pollMs, batchSize };
+ console.log(`[${new Date().toISOString()}] ct-source-ctlog starting ${ctLogs ? 'multi=[' + ctLogs + ']' : 'log=' + ctLog} poll=${pollMs}ms batch=${batchSize}`);
const s = start({
- logKey, pollMs, batchSize,
+ ...opts,
onEvent: (e) => {
const names = e.data.leaf_cert.all_domains;
- console.log(` cert#${e.data.cert_index} ${names.slice(0, 3).join(', ')}${names.length > 3 ? ` …+${names.length - 3}` : ''}`);
+ const src = e.data.log_source || '?';
+ console.log(` cert#${e.data.cert_index} [${src}] ${names.slice(0, 2).join(', ')}${names.length > 2 ? ` …+${names.length - 2}` : ''}`);
},
onStatus: (st) => {
- if (st.type === 'tick') console.log(` [tick] tree=${st.treeSize} batch=${st.fetched} stats=${JSON.stringify(st.stats)}`);
- else console.log(` [${st.type}] ${JSON.stringify(st)}`);
+ if (st.type === 'tick') console.log(` [tick ${st.log}] tree=${st.treeSize} batch=${st.fetched} x509=${st.stats.x509} pre=${st.stats.precert}`);
+ else console.log(` [${st.type}] ${JSON.stringify(st).slice(0, 220)}`);
},
});
process.on('SIGINT', () => { s.stop(); process.exit(0); });
diff --git a/dashboard.js b/dashboard.js
index eb24097..dd01750 100644
--- a/dashboard.js
+++ b/dashboard.js
@@ -24,6 +24,7 @@ const CERTSTREAM_URL = process.env.CERTSTREAM_URL || 'wss://certstream.calidog.i
// CT_SOURCE=certstream → original WebSocket flow (kept as fallback)
const CT_SOURCE = process.env.CT_SOURCE || 'ctlog';
const CT_LOG = process.env.CT_LOG || 'argon';
+const CT_LOGS = (process.env.CT_LOGS || 'argon,xenon,wyvern').split(',').map((s) => s.trim()).filter(Boolean);
const CTLOG_POLL_MS = parseInt(process.env.CTLOG_POLL_MS || '5000', 10);
const CTLOG_BATCH = parseInt(process.env.CTLOG_BATCH || '32', 10);
const DATA_DIR = path.join(__dirname, 'data');
@@ -116,17 +117,18 @@ function connectCertstream() {
}
function connectCtLog() {
- console.log(`[${new Date().toISOString()}] CT_SOURCE=ctlog log=${CT_LOG} poll=${CTLOG_POLL_MS}ms batch=${CTLOG_BATCH}`);
+ const opts = CT_LOGS.length > 1
+ ? { logKeys: CT_LOGS, pollMs: CTLOG_POLL_MS, batchSize: CTLOG_BATCH }
+ : { logKey: CT_LOG, pollMs: CTLOG_POLL_MS, batchSize: CTLOG_BATCH };
+ console.log(`[${new Date().toISOString()}] CT_SOURCE=ctlog ${CT_LOGS.length > 1 ? 'multi=[' + CT_LOGS.join(',') + ']' : 'log=' + CT_LOG} poll=${CTLOG_POLL_MS}ms batch=${CTLOG_BATCH}`);
ctLogSource.start({
- logKey: CT_LOG,
- pollMs: CTLOG_POLL_MS,
- batchSize: CTLOG_BATCH,
+ ...opts,
onEvent: handleCertEvent,
onStatus: (st) => {
if (st.type === 'sth_err' || st.type === 'fetch_err' || st.type === 'error') {
- console.error(`[${new Date().toISOString()}] ctlog ${st.type}: ${st.msg || JSON.stringify(st)}`);
+ console.error(`[${new Date().toISOString()}] ctlog ${st.type} on ${st.log || '?'}: ${st.msg || JSON.stringify(st)}`);
} else if (st.type === 'seeded') {
- console.log(`[${new Date().toISOString()}] ctlog seeded at tree_size=${st.position}`);
+ console.log(`[${new Date().toISOString()}] ctlog seeded ${st.log} at tree_size=${st.position}`);
}
},
});
← 2c5c65f ct-source-ctlog: parse PRECERT_ENTRY (TBSCertificate) for SA
·
back to Domain Sniper
·
ct-source: expand default fan-out to 6 logs (sphinx+elephant da97978 →