feat: autoconfigure kratos-changefeed

GitOrigin-RevId: 8e684d3c1ed528798c0c81cc4330858c54a39acf

Author: alnr

go.mod

@@ -15,7 +15,6 @@ replace (
 	// official SDK, allowing for the Ory CLI to consume Ory Kratos' CLI commands.
 	github.com/ory/client-go => ./internal/client-go
 	github.com/ory/x => ./oryx
-
 )
 
 require (
@@ -24,7 +23,6 @@ require (
 	github.com/arbovm/levenshtein v0.0.0-20160628152529-48b4e1c0c4d0
 	github.com/bradleyjkemp/cupaloy/v2 v2.8.0
 	github.com/bwmarrin/discordgo v0.28.1
-	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/coreos/go-oidc/v3 v3.11.0
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
 	github.com/dghubble/oauth1 v0.7.3
@@ -99,6 +97,8 @@ require (
 	google.golang.org/grpc v1.74.2
 )
 
+require github.com/cenkalti/backoff v2.2.1+incompatible
+
 require (
 	filippo.io/edwards25519 v1.1.0 // indirect
 	github.com/a8m/envsubst v1.4.2 // indirect

oryx/ipx/cidr.go

@@ -0,0 +1,23 @@
+// Copyright © 2025 Ory Corp
+// SPDX-License-Identifier: Apache-2.0
+
+package ipx
+
+import (
+	"iter"
+	"net/netip"
+)
+
+func Hosts(prefix netip.Prefix) iter.Seq[netip.Addr] {
+	prefix = prefix.Masked()
+	return func(yield func(netip.Addr) bool) {
+		if !prefix.IsValid() {
+			return
+		}
+		for addr := prefix.Addr().Next(); prefix.Contains(addr); addr = addr.Next() {
+			if !yield(addr) {
+				return
+			}
+		}
+	}
+}

oryx/logrusx/helper.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"net/http"
 	"net/url"
 	"reflect"
@@ -276,3 +277,17 @@ func (l *Logger) PopLogger(lvl logging.Level, s string, args ...interface{}) {
 		l.WithField("source", "pop").Logf(level, s, args...)
 	}
 }
+
+func (l *Logger) StdLogger(lvl logrus.Level) *log.Logger {
+	return log.New(writer{l.Logger, lvl}, "", 0)
+}
+
+type writer struct {
+	l   *logrus.Logger
+	lvl logrus.Level
+}
+
+func (w writer) Write(p []byte) (n int, err error) {
+	w.l.Log(w.lvl, strings.TrimSuffix(string(p), "\n"))
+	return len(p), nil
+}

oryx/logrusx/logrus.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"os"
 	"strings"
+	"testing"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -52,7 +53,8 @@ const ConfigSchemaID = "ory://logging-config"
 // The interface is specified instead of `jsonschema.Compiler` to allow the use of any jsonschema library fork or version.
 func AddConfigSchema(c interface {
 	AddResource(url string, r io.Reader) error
-}) error {
+},
+) error {
 	return c.AddResource(ConfigSchemaID, bytes.NewBufferString(ConfigSchema))
 }
 
@@ -233,10 +235,32 @@ func New(name string, version string, opts ...Option) *Logger {
 			return o.c.Strings("log.additional_redacted_headers")
 		}()),
 		Entry: newLogger(o.l, o).WithFields(logrus.Fields{
-			"audience": "application", "service_name": name, "service_version": version}),
+			"audience": "application", "service_name": name, "service_version": version,
+		}),
 	}
 }
 
+func NewT(t testing.TB, opts ...Option) *Logger {
+	opts = append(opts, LeakSensitive(), WithExitFunc(func(code int) {
+		t.Fatalf("Logger exited with code %d", code)
+	}))
+	l := New(t.Name(), "test", opts...)
+	l.Logger.Out = &testOutput{t}
+	return l
+}
+
+type testOutput struct {
+	t testing.TB
+}
+
+func (t *testOutput) Write(p []byte) (n int, err error) {
+	if t.t == nil {
+		return os.Stdout.Write(p)
+	}
+	t.t.Log(t.t.Name() + " " + string(p))
+	return len(p), nil
+}
+
 func NewAudit(name string, version string, opts ...Option) *Logger {
 	return New(name, version, opts...).WithField("audience", "audit")
 }

oryx/popx/db_columns.go

@@ -1,3 +1,6 @@
+// Copyright © 2025 Ory Corp
+// SPDX-License-Identifier: Apache-2.0
+
 package popx
 
 import (

oryx/popx/migrator.go

@@ -15,11 +15,11 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach-go/v2/crdb"
-	"github.com/ory/pop/v6"
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 
+	"github.com/ory/pop/v6"
 	"github.com/ory/x/cmdx"
 	"github.com/ory/x/logrusx"
 	"github.com/ory/x/otelx"

oryx/reqlog/middleware.go

@@ -102,6 +102,18 @@ func (m *Middleware) ExcludePaths(paths ...string) *Middleware {
 	return m
 }
 
+func (m *Middleware) Wrap(handler http.Handler) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
+		m.ServeHTTP(rw, r, handler.ServeHTTP)
+	})
+}
+
+func (m *Middleware) WrapFunc(handler http.HandlerFunc) http.HandlerFunc {
+	return func(rw http.ResponseWriter, r *http.Request) {
+		m.ServeHTTP(rw, r, handler)
+	}
+}
+
 func (m *Middleware) ServeHTTP(rw http.ResponseWriter, r *http.Request, next http.HandlerFunc) {
 	if m.Before == nil {
 		m.Before = DefaultBefore

oryx/tlsx/cert.go

@@ -21,6 +21,7 @@ import (
 	"math/big"
 	"os"
 	"path/filepath"
+	"slices"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -222,8 +223,8 @@ func PublicKey(key crypto.PrivateKey) interface{ Equal(x crypto.PublicKey) bool
 }
 
 // CreateSelfSignedTLSCertificate creates a self-signed TLS certificate.
-func CreateSelfSignedTLSCertificate(key interface{}) (*tls.Certificate, error) {
-	c, err := CreateSelfSignedCertificate(key)
+func CreateSelfSignedTLSCertificate(key interface{}, opts ...CertificateOpts) (*tls.Certificate, error) {
+	c, err := CreateSelfSignedCertificate(key, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -244,7 +245,7 @@ func CreateSelfSignedTLSCertificate(key interface{}) (*tls.Certificate, error) {
 }
 
 // CreateSelfSignedCertificate creates a self-signed x509 certificate.
-func CreateSelfSignedCertificate(key interface{}) (cert *x509.Certificate, err error) {
+func CreateSelfSignedCertificate(key interface{}, opts ...CertificateOpts) (cert *x509.Certificate, err error) {
 	serialNumberLimit := new(big.Int).Lsh(big.NewInt(1), 128)
 	serialNumber, err := rand.Int(rand.Reader, serialNumberLimit)
 	if err != nil {
@@ -263,14 +264,16 @@ func CreateSelfSignedCertificate(key interface{}) (cert *x509.Certificate, err e
 		},
 		NotBefore:             time.Now().UTC(),
 		NotAfter:              time.Now().UTC().Add(time.Hour * 24 * 31),
-		KeyUsage:              x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
-		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		KeyUsage:              x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
 		BasicConstraintsValid: true,
+		IsCA:                  true,
+		DNSNames:              []string{"localhost"},
+	}
+	for _, opt := range opts {
+		opt(certificate)
 	}
 
-	certificate.IsCA = true
-	certificate.KeyUsage |= x509.KeyUsageCertSign
-	certificate.DNSNames = append(certificate.DNSNames, "localhost")
 	der, err := x509.CreateCertificate(rand.Reader, certificate, certificate, PublicKey(key), key)
 	if err != nil {
 		return cert, errors.Errorf("failed to create certificate: %s", err)
@@ -292,6 +295,61 @@ func PEMBlockForKey(key interface{}) (*pem.Block, error) {
 	return &pem.Block{Type: "PRIVATE KEY", Bytes: b}, nil
 }
 
+// NewClientCert creates a new client TLS certificate signed by the given CA.
+func NewClientCert(CAcert *x509.Certificate, CAkey crypto.PrivateKey, opts ...CertificateOpts) (*tls.Certificate, error) {
+	if !slices.Contains(CAcert.ExtKeyUsage, x509.ExtKeyUsageClientAuth) {
+		return nil, errors.Errorf("the CA certificate does not have the client authentication extended key usage (OID 1.3.6.1.5.5.7.3.2) set")
+	}
+	serialNumberLimit := new(big.Int).Lsh(big.NewInt(1), 128)
+	serialNumber, err := rand.Int(rand.Reader, serialNumberLimit)
+	if err != nil {
+		return nil, errors.Errorf("failed to generate serial number: %s", err)
+	}
+
+	key, err := rsa.GenerateKey(rand.Reader, 3072)
+	if err != nil {
+		return nil, errors.Errorf("failed to generate private key: %s", err)
+	}
+
+	template := &x509.Certificate{
+		SerialNumber: serialNumber,
+		Subject: pkix.Name{
+			Organization: []string{"Ory GmbH"},
+			CommonName:   "ORY",
+		},
+		Issuer:                CAcert.Subject,
+		NotBefore:             time.Now().UTC(),
+		NotAfter:              CAcert.NotAfter,
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
+		BasicConstraintsValid: true,
+		IsCA:                  false,
+	}
+	for _, opt := range opts {
+		opt(template)
+	}
+
+	der, err := x509.CreateCertificate(rand.Reader, template, CAcert, PublicKey(key), CAkey)
+	if err != nil {
+		return nil, errors.Errorf("failed to create certificate: %s", err)
+	}
+
+	pemCert := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
+	pemBlock, err := PEMBlockForKey(key)
+	if err != nil {
+		return nil, err
+	}
+	pemKey := pem.EncodeToMemory(pemBlock)
+
+	cert, err := tls.X509KeyPair(pemCert, pemKey)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return &cert, nil
+}
+
+type CertificateOpts func(*x509.Certificate)
+
 // CreateSelfSignedCertificateForTest writes a new, self-signed TLS
 // certificate+key (in PEM format) to a temporary location on disk and returns
 // the paths to both, and the respective contents in base64 encoding. The