Feature/valkey json notifications

cmd/certspotter/main.go

@@ -26,6 +26,8 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/valkey-io/valkey-go"
+
 	"software.sslmate.com/src/certspotter/ctclient"
 	"software.sslmate.com/src/certspotter/loglist"
 	"software.sslmate.com/src/certspotter/monitor"
@@ -109,6 +111,27 @@ func defaultScriptDir() string {
 func defaultEmailFile() string {
 	return filepath.Join(defaultConfigDir(), "email_recipients")
 }
+func defaultValkeyUrl() string {
+	if envVar := os.Getenv("CERTSPOTTER_VALKEY_URL"); envVar != "" {
+		return envVar
+	} else {
+		return ""
+	}
+}
+func defaultValkeyStream() string {
+	if envVar := os.Getenv("CERTSPOTTER_VALKEY_STREAM"); envVar != "" {
+		return envVar
+	} else {
+		return "certspotter"
+	}
+}
+func defaultValkeyStreamThreshold() string {
+	if envVar := os.Getenv("CERTSPOTTER_VALKEY_STREAM_THRESHOLD"); envVar != "" {
+		return envVar
+	} else {
+		return "10000"
+	}
+}
 
 func simplifyError(err error) error {
 	var pathErr *fs.PathError
@@ -161,17 +184,20 @@ func main() {
 	loglist.UserAgent = ctclient.UserAgent
 
 	var flags struct {
-		email       []string
-		healthcheck time.Duration
-		logs        string
-		noSave      bool
-		script      string
-		startAtEnd  bool
-		stateDir    string
-		stdout      bool
-		verbose     bool
-		version     bool
-		watchlist   string
+		email                 []string
+		healthcheck           time.Duration
+		logs                  string
+		noSave                bool
+		script                string
+		startAtEnd            bool
+		stateDir              string
+		stdout                bool
+		verbose               bool
+		version               bool
+		watchlist             string
+		valkeyUrl             string
+		valkeyStream          string
+		valkeyStreamThreshold string
 	}
 	flag.Func("email", "Email address to contact when matching certificate is discovered (repeatable)", appendFunc(&flags.email))
 	flag.DurationVar(&flags.healthcheck, "healthcheck", 24*time.Hour, "How frequently to perform a health check")
@@ -184,6 +210,9 @@ func main() {
 	flag.BoolVar(&flags.verbose, "verbose", false, "Print detailed information about certspotter's operation to stderr")
 	flag.BoolVar(&flags.version, "version", false, "Print version and exit")
 	flag.StringVar(&flags.watchlist, "watchlist", defaultWatchListPathIfExists(), "File containing domain names to watch")
+	flag.StringVar(&flags.valkeyUrl, "valkey_url", defaultValkeyUrl(), "Send JSON events to the Valkey server identified by this URL")
+	flag.StringVar(&flags.valkeyStream, "valkey_stream", defaultValkeyStream(), "Send JSON events to this Valkey stream")
+	flag.StringVar(&flags.valkeyStreamThreshold, "valkey_stream_threshold", defaultValkeyStreamThreshold(), "Trims the Valkey stream to maintain a specific size or remove old JSON events")
 	flag.Parse()
 
 	if flags.version {
@@ -195,14 +224,27 @@ func main() {
 		os.Exit(2)
 	}
 
+	var valkeyClient valkey.Client
+	var err error
+	if flags.valkeyUrl != "" {
+		valkeyClient, err = valkey.NewClient(valkey.MustParseURL(flags.valkeyUrl))
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "%s: unable to connect to Valkey: %s\n", programName, err)
+			os.Exit(1)
+		}
+	}
+
 	fsstate := &monitor.FilesystemState{
-		StateDir:  flags.stateDir,
-		CacheDir:  defaultCacheDir(),
-		SaveCerts: !flags.noSave,
-		Script:    flags.script,
-		ScriptDir: defaultScriptDir(),
-		Email:     flags.email,
-		Stdout:    flags.stdout,
+		StateDir:              flags.stateDir,
+		CacheDir:              defaultCacheDir(),
+		SaveCerts:             !flags.noSave,
+		Script:                flags.script,
+		ScriptDir:             defaultScriptDir(),
+		Email:                 flags.email,
+		Stdout:                flags.stdout,
+		ValkeyClient:          valkeyClient,
+		ValkeyStream:          flags.valkeyStream,
+		ValkeyStreamThreshold: flags.valkeyStreamThreshold,
 	}
 	config := &monitor.Config{
 		LogListSource:       flags.logs,
@@ -221,14 +263,15 @@ func main() {
 		os.Exit(1)
 	}
 
-	if len(fsstate.Email) == 0 && !emailFileExists && fsstate.Script == "" && !fileExists(fsstate.ScriptDir) && fsstate.Stdout == false {
+	if len(fsstate.Email) == 0 && !emailFileExists && fsstate.Script == "" && !fileExists(fsstate.ScriptDir) && fsstate.Stdout == false && fsstate.ValkeyClient == nil {
 		fmt.Fprintf(os.Stderr, "%s: no notification methods were specified\n", programName)
 		fmt.Fprintf(os.Stderr, "Please specify at least one of the following notification methods:\n")
 		fmt.Fprintf(os.Stderr, " - Place one or more email addresses in %s (one address per line)\n", defaultEmailFile())
 		fmt.Fprintf(os.Stderr, " - Place one or more executable scripts in the %s directory\n", fsstate.ScriptDir)
 		fmt.Fprintf(os.Stderr, " - Specify an email address using the -email flag\n")
 		fmt.Fprintf(os.Stderr, " - Specify the path to an executable script using the -script flag\n")
 		fmt.Fprintf(os.Stderr, " - Specify the -stdout flag\n")
+		fmt.Fprintf(os.Stderr, " - Specify the -valkey_url flag\n")
 		os.Exit(2)
 	}
 
@@ -252,7 +295,7 @@ func main() {
 	defer stop()
 
 	go func() {
-		ticker := time.NewTicker(24*time.Hour)
+		ticker := time.NewTicker(24 * time.Hour)
 		defer ticker.Stop()
 		for {
 			fsstate.PruneOldErrors()

go.mod

@@ -8,6 +8,10 @@ require (
 	golang.org/x/sync v0.15.0
 )
 
-require golang.org/x/text v0.26.0 // indirect
+require (
+	github.com/valkey-io/valkey-go v1.0.66 // indirect
+	golang.org/x/sys v0.33.0 // indirect
+	golang.org/x/text v0.26.0 // indirect
+)
 
 retract v0.19.0 // Contains serious bugs.

go.sum

@@ -1,8 +1,12 @@
+github.com/valkey-io/valkey-go v1.0.66 h1:DIEF1XpwbO78xK2sMTghYE3Bz6pePWJTNxKtgoAuA3A=
+github.com/valkey-io/valkey-go v1.0.66/go.mod h1:bHmwjIEOrGq/ubOJfh5uMRs7Xj6mV3mQ/ZXUbmqpjqY=
 golang.org/x/crypto v0.39.0 h1:SHs+kF4LP+f+p14esP5jAoDpHU8Gu/v9lFRK6IT5imM=
 golang.org/x/crypto v0.39.0/go.mod h1:L+Xg3Wf6HoL4Bn4238Z6ft6KfEpN0tJGo53AAPC632U=
 golang.org/x/net v0.41.0 h1:vBTly1HeNPEn3wtREYfy4GZ/NECgw2Cnl+nK6Nz3uvw=
 golang.org/x/net v0.41.0/go.mod h1:B/K4NNqkfmg07DQYrbwvSluqCJOOXwUjeb/5lOisjbA=
 golang.org/x/sync v0.15.0 h1:KWH3jNZsfyT6xfAfKiz6MRNmd46ByHDYaZ7KSkCtdW8=
 golang.org/x/sync v0.15.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
+golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/text v0.26.0 h1:P42AVeLghgTYr4+xUnTRKDMqpar+PtX7KWuNQL21L8M=
 golang.org/x/text v0.26.0/go.mod h1:QK15LZJUUQVJxhz7wXgxSy/CJaTFjd0G+YLonydOVQA=

monitor/discoveredcert.go

@@ -57,10 +57,15 @@ func (cert *DiscoveredCert) pemChain() []byte {
 
 func (cert *DiscoveredCert) json() any {
 	object := map[string]any{
+		"log_uri":       cert.LogEntry.Log.GetMonitoringURL(),
+		"entry_index":   fmt.Sprint(cert.LogEntry.Index),
+		"watch_item":    cert.WatchItem.String(),
 		"tbs_sha256":    hex.EncodeToString(cert.TBSSHA256[:]),
+		"cert_sha256":   hex.EncodeToString(cert.SHA256[:]),
 		"pubkey_sha256": hex.EncodeToString(cert.PubkeySHA256[:]),
-		"dns_names":     cert.Identifiers.DNSNames,
-		"ip_addresses":  cert.Identifiers.IPAddrs,
+
+		"dns_names":    cert.Identifiers.DNSNames,
+		"ip_addresses": cert.Identifiers.IPAddrs,
 	}
 
 	if cert.Info.ValidityParseError == nil {
@@ -71,6 +76,24 @@ func (cert *DiscoveredCert) json() any {
 		object["not_after"] = nil
 	}
 
+	if cert.Info.SubjectParseError == nil {
+		object["subject_dn"] = cert.Info.Subject.String()
+	} else {
+		object["subject_dn"] = nil
+	}
+
+	if cert.Info.IssuerParseError == nil {
+		object["issuer_dn"] = cert.Info.Issuer.String()
+	} else {
+		object["issuer_dn"] = nil
+	}
+
+	if cert.Info.SerialNumberParseError == nil {
+		object["serial"] = fmt.Sprintf("%x", cert.Info.SerialNumber)
+	} else {
+		object["serial"] = nil
+	}
+
 	return object
 }
 
@@ -87,6 +110,59 @@ func writeCertFiles(cert *DiscoveredCert, paths *certPaths) error {
 	return nil
 }
 
+func certNotificationJson(cert *DiscoveredCert) any {
+	object := map[string]any{
+		"event":   "discovered_cert",
+		"summary": certNotificationSummary(cert),
+
+		"log_uri":       cert.LogEntry.Log.GetMonitoringURL(),
+		"entry_index":   fmt.Sprint(cert.LogEntry.Index),
+		"watch_item":    cert.WatchItem.String(),
+		"tbs_sha256":    hex.EncodeToString(cert.TBSSHA256[:]),
+		"cert_sha256":   hex.EncodeToString(cert.SHA256[:]),
+		"pubkey_sha256": hex.EncodeToString(cert.PubkeySHA256[:]),
+
+		"dns_names":    cert.Identifiers.DNSNames,
+		"ip_addresses": cert.Identifiers.IPAddrs,
+	}
+
+	if cert.Info.ValidityParseError == nil {
+		object["not_before"] = cert.Info.Validity.NotBefore
+		object["not_after"] = cert.Info.Validity.NotAfter
+	} else {
+		object["not_before"] = nil
+		object["not_after"] = nil
+		object["validity_parse_error"] = cert.Info.ValidityParseError.Error()
+	}
+
+	if cert.Info.SubjectParseError == nil {
+		object["subject_dn"] = cert.Info.Subject.String()
+	} else {
+		object["subject_dn"] = nil
+		object["subject_parser_error"] = cert.Info.SubjectParseError.Error()
+	}
+
+	if cert.Info.IssuerParseError == nil {
+		object["issuer_dn"] = cert.Info.Issuer.String()
+	} else {
+		object["issuer_dn"] = nil
+		object["issuer_parser_error"] = cert.Info.IssuerParseError.Error()
+	}
+
+	if cert.Info.SerialNumberParseError == nil {
+		object["serial"] = fmt.Sprintf("%x", cert.Info.SerialNumber)
+	} else {
+		object["serial"] = nil
+		object["serial_parse_error"] = cert.Info.SerialNumberParseError.Error()
+	}
+
+	if cert.ChainError != nil {
+		object["chain_error"] = cert.ChainError.Error()
+	}
+
+	return object
+}
+
 func certNotificationEnviron(cert *DiscoveredCert, paths *certPaths) []string {
 	env := []string{
 		"EVENT=discovered_cert",

monitor/fsstate.go

@@ -23,6 +23,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/valkey-io/valkey-go"
+
 	"software.sslmate.com/src/certspotter/cttypes"
 	"software.sslmate.com/src/certspotter/loglist"
 	"software.sslmate.com/src/certspotter/merkletree"
@@ -32,14 +34,17 @@ const keepErrorDays = 7
 const errorDateFormat = "2006-01-02"
 
 type FilesystemState struct {
-	StateDir  string
-	CacheDir  string
-	SaveCerts bool
-	Script    string
-	ScriptDir string
-	Email     []string
-	Stdout    bool
-	errorMu   sync.Mutex
+	StateDir              string
+	CacheDir              string
+	SaveCerts             bool
+	Script                string
+	ScriptDir             string
+	Email                 []string
+	Stdout                bool
+	errorMu               sync.Mutex
+	ValkeyClient          valkey.Client
+	ValkeyStream          string
+	ValkeyStreamThreshold string
 }
 
 func (s *FilesystemState) logStateDir(logID LogID) string {
@@ -163,6 +168,7 @@ func (s *FilesystemState) NotifyCert(ctx context.Context, cert *DiscoveredCert)
 	}
 
 	if err := s.notify(ctx, &notification{
+		json:    certNotificationJson(cert),
 		summary: certNotificationSummary(cert),
 		environ: certNotificationEnviron(cert, paths),
 		text:    certNotificationText(cert, paths),
@@ -203,6 +209,15 @@ func (s *FilesystemState) NotifyMalformedEntry(ctx context.Context, entry *LogEn
 		return fmt.Errorf("error saving texT file: %w", err)
 	}
 
+	json := map[string]any{
+		"event":       "malformed_cert",
+		"summary":     summary,
+		"log_uri":     entry.Log.GetMonitoringURL(),
+		"entry_index": fmt.Sprint(entry.Index),
+		"leaf_hash":   leafHash.Base64String(),
+		"parse_error": parseError.Error(),
+	}
+
 	environ := []string{
 		"EVENT=malformed_cert",
 		"SUMMARY=" + summary,
@@ -216,6 +231,7 @@ func (s *FilesystemState) NotifyMalformedEntry(ctx context.Context, entry *LogEn
 	}
 
 	if err := s.notify(ctx, &notification{
+		json:    json,
 		environ: environ,
 		summary: summary,
 		text:    text.String(),
@@ -242,6 +258,10 @@ func (s *FilesystemState) errorDir(ctlog *loglist.Log) string {
 
 func (s *FilesystemState) NotifyHealthCheckFailure(ctx context.Context, ctlog *loglist.Log, info HealthCheckFailure) error {
 	textPath := filepath.Join(s.healthCheckDir(ctlog), healthCheckFilename())
+	json := map[string]any{
+		"event":   "error",
+		"summary": info.Summary(),
+	}
 	environ := []string{
 		"EVENT=error",
 		"SUMMARY=" + info.Summary(),
@@ -252,6 +272,7 @@ func (s *FilesystemState) NotifyHealthCheckFailure(ctx context.Context, ctlog *l
 		return fmt.Errorf("error saving text file: %w", err)
 	}
 	if err := s.notify(ctx, &notification{
+		json:    json,
 		environ: environ,
 		summary: info.Summary(),
 		text:    text,

monitor/notify.go

@@ -12,6 +12,7 @@ package monitor
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io/fs"
@@ -26,6 +27,7 @@ import (
 var stdoutMu sync.Mutex
 
 type notification struct {
+	json    any
 	environ []string
 	summary string
 	text    string
@@ -54,6 +56,16 @@ func (s *FilesystemState) notify(ctx context.Context, notif *notification) error
 		}
 	}
 
+	if s.ValkeyClient != nil {
+		notifJson, err := json.Marshal(notif.json)
+		if err != nil {
+			return err
+		}
+		if err := s.ValkeyClient.Do(ctx, s.ValkeyClient.B().Xadd().Key(s.ValkeyStream).Maxlen().Almost().Threshold(s.ValkeyStreamThreshold).Id("*").FieldValue().FieldValue("data", string(notifJson)).Build()).Error(); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }