lixen/logwisp
SHA256
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.28.0 option schema in struct tags, schema replacing plugin factory metadata, tui prep

Browse Source
This commit is contained in:
lixen committed 2026-10-09 14:43:44 -04:00
1 parent a1648ce17b
commit e8e5f05e47
43 files changed
+1345 -1211
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

AGENTS.md
+6 -3
View File
@@ -73,9 +73,12 @@ change touches.
## LogWisp
- Plugin options decode through `config.Scan`, never `lconfig.ScanMap`: unknown
keys must fail at any depth. Set defaults on the options struct before the
call; `TestEveryPluginRejectsUnknownKeys` covers every registered plugin.
- Plugin options are a schema in `internal/config`: struct tags (`default:`,
`help:`, `lw:` rules) and one catalogue row per plugin type, without which
`RegisterSource`/`RegisterSink` fail. Constructors decode through
`config.Decode`, which applies the defaults, rules and pure checks and
refuses unknown keys at any depth; they keep only I/O. Rules that read
nothing belong in a `Check` method, so `ValidateConfig` sees them offline.
- `cmd/lw/cli.go` owns the command line: lw's own flags, and the commands
table whose groups (`auth`, `tls`, `preset`, `config`) share one flag
mechanism; `config.Load` takes the parsed `config.Args`. Presets are rows of
cmd/lw/auth.go
+1 -1
View File
@@ -258,7 +258,7 @@ func dialPolicy(o *config.TLSOptions, host, user, passwordFile string, transport
return nil, nil, err
}
policy, err := authz.New(&config.AuthOptions{Type: authz.MethodSCRAM, Username: user, PasswordFile: passwordFile},
tlsCfg, nil, authz.RoleDialer, transport)
tlsCfg, nil, config.Dialer, transport)
if err != nil {
return nil, nil, err
}
cmd/lw/reload_test.go
+4 -18
View File
@@ -117,22 +117,6 @@ func TestEveryPluginRejectsUnknownKeys(t *testing.T) {
}
}
// A misspelled spec key reaches the plugin as the nested key it names, so the
// sink refuses to build instead of starting without the TLS it was asked for.
func TestMisspelledSpecKeyFailsPluginConstruction(t *testing.T) {
testLogger(t)
path := filepath.Join(t.TempDir(), "empty.toml")
testutil.WriteFile(t, path, "")
cfg := loadTestConfig(t, path, "--source", "null", "--sink", "http,host=127.0.0.1,port=15861,tls.enabeld=true")
svc, err := bootstrapService(context.Background(), cfg)
if err == nil {
svc.Shutdown()
}
if err == nil || !strings.Contains(err.Error(), `unknown key "tls.enabeld"`) {
t.Fatalf("misspelled spec key accepted: %v", err)
}
}
// lw --check builds every plugin, so a bad option fails it, but starts none:
// the port a valid sink names stays free and its log directory uncreated.
func TestCheckBuildsWithoutStarting(t *testing.T) {
@@ -152,7 +136,9 @@ func TestCheckBuildsWithoutStarting(t *testing.T) {
if _, err := os.Stat(out); !errors.Is(err, fs.ErrNotExist) {
t.Fatalf("check created the file sink's directory: %v", err)
}
if code := checkConfig(loadTestConfig(t, path, "--source", "null", "--sink", "http,port=15862,tls.enabled=true")); code != 1 {
t.Fatalf("listener TLS without a certificate: exit %d, want 1", code)
missing := filepath.Join(dir, "missing")
if code := checkConfig(loadTestConfig(t, path, "--source", "null", "--sink",
"http,port=15862,tls.enabled=true,tls.cert_file="+missing+",tls.key_file="+missing)); code != 1 {
t.Fatalf("listener TLS without its certificate file: exit %d, want 1", code)
}
}
doc/architecture.md
+7 -4
View File
@@ -55,11 +55,14 @@ Package map:
Every plugin registers itself in an `init()` function, and
`cmd/lw/bootstrap.go` blank-imports each package to trigger those
`init()`s. Adding a plugin therefore means writing the package, calling
`plugin.RegisterSource` / `plugin.RegisterSink`, and adding one blank import.
`plugin.RegisterSource` / `plugin.RegisterSink`, adding one blank import, and
adding the type's row to the catalogue in `internal/config/schema.go`, which
registration requires.
Registration may attach metadata. The `console` source declares
`MaxInstances: 1`, because a process has only one stdin; the per-pipeline
registry rejects a second instance of any such type.
The row holds the type's options struct, whose tags declare its defaults and
rules, and whether the type is single. The `console` source is single,
because a process has only one stdin; loading a configuration with a second
instance of a single type fails, naming both pipelines.
## Data Flow
doc/todo.md
+6 -29
View File
@@ -85,6 +85,11 @@ What exists:
(`internal/config/config.go`), presets as rows with typed parameters
(`preset.go`), the command-line and environment grammar (`spec.go`),
`ValidateConfig`, `--check` and `--dump`.
- The option schema, in `internal/config`: `default:`, `help:` and `lw:` tags
on the option structs, one catalogue row per plugin type, `Decode` (every
constructor decodes through it and keeps only I/O), `Coerce` for typed
command-line values, and a `ValidateConfig` that decodes every plugin
offline and names the refused key's path.
- `internal/config` already compiles for `GOOS=js`; it links no net/http,
crypto/tls or terminal code.
- `github.com/lixenwraith/terminal` (pinned) and its `tui` package: regions,
@@ -93,41 +98,13 @@ What exists:
blending, RGBTo256, RGBTo16.
What is missing:
- A machine-readable option schema. Defaults, enums and required checks live
inside the plugin constructors (about 50 `Default*` constants, 24 inline
validators); `doc/sources.md`, `doc/sinks.md`, `config/logwisp.toml` and the
site's catalogue copy them by hand, and the site's copy already lags.
- Offline plugin validation: the constructors mix their checks with reading
certificates, credentials and files.
- Typed command-line values: `lw --sink http,port=8080 --dump` prints
`port = "8080"`, a file `port = 8080`.
- A host-independent preset: `fileSource` (`preset.go`) decides directory or
pattern with `os.Stat`, so off the target host `path=/var/log` expands to
directory `/var`, pattern `log`.
- Emitters: only parsers exist for the three forms.
Steps:
1. Schema, in `internal/config`, as the one source of truth.
- Tags on the option structs: `default:`, `help:`, and `lw:` for required,
enum, min and max, what zero means (default, none, unlimited), hints
(path, regex, file) and roles (listener, dialer) on the shared tls, auth
and acl tables.
- One catalogue table, one row per plugin type: role, network role, single
instance, summary, option struct. `RegisterSource`/`RegisterSink` panic on
a type without a row.
- `Decode[T](role, type, map)` applies defaults, unknown-key and tag checks,
then the struct's pure `Check()`. Every constructor decodes through it and
keeps only I/O; the `Default*` constants and inline validators go.
- Pure cross-field checks become methods that tlsx, authz and netacl call
first: TLS role rules, auth bindings without file reads, ACL parsing, the
http sink's paths, filter regexes.
- `Coerce` converts command-line and environment strings, and JSON floats,
to the declared kinds; `spec.go` uses it, so a command line dumps typed.
- `ValidateConfig` adds unique plugin ids per pipeline and decodes every
plugin, naming the path (`pipelines[0].plugin_sinks[view].config.port`).
It opens nothing; `lw --check` stays the authority on files.
- Keep under test what moves: file sink `min_disk_free_mb` (negative means
100, 0 means none), a backoff maximum below the minimum, format `text`.
1. Schema, in `internal/config`: done.
2. Engine, `internal/compose`.
- Imports internal/config, internal/version and lixenwraith/toml only; a
test fails when `go list -deps` shows net, crypto/tls, os/exec, os/signal,
internal/authz/authz.go
+21 -113
View File
@@ -7,6 +7,7 @@ package authz
import (
"bufio"
"cmp"
"context"
"crypto/tls"
"errors"
@@ -42,21 +43,6 @@ const (
BindingForce = "force"
)
// Role selects the validation and behavior appropriate to the call site
type Role int
const (
// RoleListener authenticates peers of a plugin with no node concept: the
// tcp and http sinks
RoleListener Role = iota
// RoleChainListener also binds the node label a peer declares: the
// tcp_chain and http_chain sources
RoleChainListener
// RoleDialer proves itself to, or pins, the server: the tcp_chain and
// http_chain sinks
RoleDialer
)
// Transport tells HTTP plugins, which carry bearer tokens, from TCP ones
type Transport int
@@ -72,7 +58,7 @@ var ErrRefused = errors.New("auth: refused")
// Policy is the compiled form of config.AuthOptions
type Policy struct {
role Role
side config.Side
transport Transport
method string
identity string // mtls: identity field; scram: certificate-to-user binding, "" = none
@@ -113,38 +99,25 @@ func (id Identity) Apply(meta map[string]any) {
// tlsCfg is what tlsx built from the sibling `tls` block: a policy the
// transport cannot enforce is rejected here rather than silently accepted.
// acl is netacl's from the `acl` block, nil for a dialer.
func New(o *config.AuthOptions, tlsCfg *tls.Config, acl *netacl.Policy, role Role, transport Transport) (*Policy, error) {
if o == nil {
return nil, nil
func New(o *config.AuthOptions, tlsCfg *tls.Config, acl *netacl.Policy, side config.Side, transport Transport) (*Policy, error) {
state := config.TLSState{On: tlsCfg != nil}
if tlsCfg != nil {
state.ClientAuth = tlsCfg.ClientAuth == tls.RequireAndVerifyClientCert
state.Unverified = tlsCfg.InsecureSkipVerify && tlsCfg.VerifyPeerCertificate == nil // a pin verifies there
}
switch o.Type {
case "", MethodNone:
// Tuning keys are ignored with the block, but one that names peers or
// credentials means auth was intended and the type was forgotten
if key := intentKey(o); key != "" {
return nil, fmt.Errorf("auth: %s is set but auth.type is %q", key, MethodNone)
}
return nil, nil
case MethodMTLS, MethodSCRAM:
default:
return nil, fmt.Errorf("auth: type %q (valid: %q, %q, %q)", o.Type, MethodNone, MethodMTLS, MethodSCRAM)
}
if tlsCfg == nil && (o.Type != MethodSCRAM || len(o.TrustedProxies) == 0) {
return nil, fmt.Errorf("auth: type %q requires tls.enabled", o.Type)
}
if role == RoleDialer && tlsCfg != nil && tlsCfg.InsecureSkipVerify && tlsCfg.VerifyPeerCertificate == nil {
// An unverified server makes identities claims and exposes credentials;
// tls.pin_sha256 verifies it in VerifyPeerCertificate instead
return nil, fmt.Errorf("auth: type %q cannot be used with tls.insecure_skip_verify", o.Type)
}
binding, err := nodeBinding(o.NodeBinding, role)
if err != nil {
if err := o.Check(side, transport == HTTP, state); err != nil || o == nil || o.Type == "" || o.Type == MethodNone {
return nil, err
}
p := &Policy{role: role, transport: transport, method: o.Type, binding: binding, acl: acl}
binding := BindingNone
if side == config.ChainListener {
// force is the only binding under which a misconfigured or hostile edge
// cannot mislabel its entries
binding = cmp.Or(o.NodeBinding, BindingForce)
}
p := &Policy{side: side, transport: transport, method: o.Type, binding: binding, acl: acl}
var err error
if o.Type == MethodMTLS {
err = p.compileMTLS(o, tlsCfg)
err = p.compileMTLS(o)
} else {
err = p.compileSCRAM(o, tlsCfg)
}
@@ -154,57 +127,8 @@ func New(o *config.AuthOptions, tlsCfg *tls.Config, acl *netacl.Policy, role Rol
return p, nil
}
// intentKey names the first key that only makes sense with authentication on
func intentKey(o *config.AuthOptions) string {
switch {
case len(o.Allow) > 0:
return "allow"
case len(o.AllowPatterns) > 0:
return "allow_patterns"
}
return scramKey(o)
}
func nodeBinding(binding string, role Role) (string, error) {
if role != RoleChainListener {
if binding != "" && binding != BindingNone {
return "", fmt.Errorf("auth: node_binding %q applies only to chain sources", binding)
}
return BindingNone, nil
}
switch binding {
case "":
// The only setting under which a misconfigured or hostile edge cannot
// mislabel its entries
return BindingForce, nil
case BindingNone, BindingAssert, BindingForce:
return binding, nil
}
return "", fmt.Errorf("auth: node_binding %q (valid: %q, %q, %q)", binding, BindingNone, BindingAssert, BindingForce)
}
func identityMode(mode string) (string, error) {
switch mode {
case "":
return tlsx.IdentityCN, nil
case tlsx.IdentityCN, tlsx.IdentitySANDNS, tlsx.IdentitySANURI, tlsx.IdentitySANEmail:
return mode, nil
}
return "", fmt.Errorf("auth: identity %q (valid: %q, %q, %q, %q)",
mode, tlsx.IdentityCN, tlsx.IdentitySANDNS, tlsx.IdentitySANURI, tlsx.IdentitySANEmail)
}
func (p *Policy) compileMTLS(o *config.AuthOptions, tlsCfg *tls.Config) error {
if key := scramKey(o); key != "" {
return fmt.Errorf("auth: %s applies only to type %q", key, MethodSCRAM)
}
if p.role != RoleDialer && tlsCfg.ClientAuth != tls.RequireAndVerifyClientCert {
return fmt.Errorf("auth: type %q requires tls.client_auth", MethodMTLS)
}
var err error
if p.identity, err = identityMode(o.Identity); err != nil {
return err
}
func (p *Policy) compileMTLS(o *config.AuthOptions) error {
p.identity = cmp.Or(o.Identity, tlsx.IdentityCN)
p.allow = make(map[string]struct{}, len(o.Allow))
for _, a := range o.Allow {
if a = strings.TrimSpace(a); a != "" {
@@ -221,22 +145,6 @@ func (p *Policy) compileMTLS(o *config.AuthOptions, tlsCfg *tls.Config) error {
return nil
}
func scramKey(o *config.AuthOptions) string {
switch {
case o.CredentialsFile != "":
return "credentials_file"
case o.TokenLifetimeMS != 0:
return "token_lifetime_ms"
case o.Username != "":
return "username"
case o.PasswordFile != "":
return "password_file"
case len(o.TrustedProxies) > 0:
return "trusted_proxies"
}
return ""
}
// Authorize checks the mtls certificate identity of a completed handshake. It
// refuses under scram, so a listener that skipped its exchange fails closed.
// A nil policy authorizes everything and yields the zero Identity.
@@ -566,7 +474,7 @@ func (p *Policy) LogStartup(l *log.Logger, component, id string, trustNode bool)
l.Warn("msg", w, "component", component, "instance_id", id)
}
}
if p.role == RoleDialer {
if p.side == config.Dialer {
return
}
if p.BehindProxy() && p.listener.proxy.exposedHop() {
@@ -687,7 +595,7 @@ func (p *Policy) Stats() map[string]any {
d["auth_identity"] = p.identity
d["auth_unrestricted"] = p.Unrestricted()
}
if p.role == RoleChainListener {
if p.side == config.ChainListener {
d["node_binding"] = p.binding
}
return d
internal/authz/authz_test.go
+40 -40
View File
@@ -66,7 +66,7 @@ func TestPeerIdentityModes(t *testing.T) {
func TestNewDisabled(t *testing.T) {
for _, o := range []*config.AuthOptions{nil, {}, {Type: MethodNone}} {
p, err := New(o, nil, nil, RoleListener, TCP)
p, err := New(o, nil, nil, config.Listener, TCP)
if err != nil {
t.Fatalf("New(%+v) error: %v", o, err)
}
@@ -122,41 +122,41 @@ func TestNewValidation(t *testing.T) {
name string
auth *config.AuthOptions
tls *tls.Config
role Role
role config.Side
transport Transport
want string
}{
{"unknown type", &config.AuthOptions{Type: "kerberos"}, mtlsListenerTLS(), RoleListener, TCP, `type "kerberos"`},
{"allow list without a type", &config.AuthOptions{Allow: []string{"edge-01"}}, mtlsListenerTLS(), RoleListener, TCP, "allow is set"},
{"credentials without a type", &config.AuthOptions{CredentialsFile: f.creds}, mtlsListenerTLS(), RoleListener, TCP, "credentials_file is set"},
{"no tls", &config.AuthOptions{Type: MethodMTLS}, nil, RoleListener, TCP, "requires tls.enabled"},
{"no client_auth", &config.AuthOptions{Type: MethodMTLS}, &tls.Config{}, RoleListener, TCP, "requires tls.client_auth"},
{"unknown identity", &config.AuthOptions{Type: MethodMTLS, Identity: "serial"}, mtlsListenerTLS(), RoleListener, TCP, `identity "serial"`},
{"bad pattern", &config.AuthOptions{Type: MethodMTLS, AllowPatterns: []string{"^edge-("}}, mtlsListenerTLS(), RoleListener, TCP, "allow_patterns[0]"},
{"unknown binding", &config.AuthOptions{Type: MethodMTLS, NodeBinding: "maybe"}, mtlsListenerTLS(), RoleChainListener, TCP, `node_binding "maybe"`},
{"binding on plain listener", &config.AuthOptions{Type: MethodMTLS, NodeBinding: BindingForce}, mtlsListenerTLS(), RoleListener, TCP, "only to chain sources"},
{"binding on dialer", &config.AuthOptions{Type: MethodMTLS, NodeBinding: BindingForce}, &tls.Config{}, RoleDialer, TCP, "only to chain sources"},
{"dialer skips verify", &config.AuthOptions{Type: MethodMTLS}, &tls.Config{InsecureSkipVerify: true}, RoleDialer, TCP, "insecure_skip_verify"},
{"mtls with a password", &config.AuthOptions{Type: MethodMTLS, PasswordFile: pw}, &tls.Config{}, RoleDialer, TCP, "password_file applies only"},
{"scram without tls", listener(config.AuthOptions{}), nil, RoleListener, TCP, "requires tls.enabled"},
{"scram without credentials", scram(config.AuthOptions{}), f.serverTLS, RoleListener, TCP, "requires credentials_file"},
{"scram with an allow list", listener(config.AuthOptions{Allow: []string{"edge-01"}}), f.serverTLS, RoleListener, TCP, "credentials file is the allow list"},
{"scram listener with a password", listener(config.AuthOptions{Username: "edge-01"}), f.serverTLS, RoleListener, TCP, "only to dialers"},
{"scram certificate binding without client_auth", listener(config.AuthOptions{Identity: "cn"}), f.serverTLS, RoleListener, TCP, "requires tls.client_auth"},
{"token lifetime on tcp", listener(config.AuthOptions{TokenLifetimeMS: 60000}), f.serverTLS, RoleListener, TCP, "only to HTTP listeners"},
{"token lifetime too short", listener(config.AuthOptions{TokenLifetimeMS: 900}), f.serverTLS, RoleListener, HTTP, "outside"},
{"token lifetime too long", listener(config.AuthOptions{TokenLifetimeMS: 1e13}), f.serverTLS, RoleListener, HTTP, "outside"},
{"scram listener without a certificate", listener(config.AuthOptions{}), &tls.Config{}, RoleListener, TCP, "listener certificate"},
{"scram dialer without a password", scram(config.AuthOptions{Username: "edge-01"}), &tls.Config{}, RoleDialer, TCP, "requires username and password_file"},
{"scram dialer with credentials", dialer(config.AuthOptions{CredentialsFile: f.creds}), &tls.Config{}, RoleDialer, TCP, "only to listeners"},
{"scram dialer skips verify", dialer(config.AuthOptions{}), &tls.Config{InsecureSkipVerify: true}, RoleDialer, TCP, "insecure_skip_verify"},
{"trusted proxies without a type", &config.AuthOptions{TrustedProxies: proxies}, nil, RoleListener, HTTP, "trusted_proxies is set"},
{"trusted proxies under mtls", &config.AuthOptions{Type: MethodMTLS, TrustedProxies: proxies}, mtlsListenerTLS(), RoleListener, HTTP, "trusted_proxies applies only"},
{"trusted proxies on a chain source", listener(config.AuthOptions{TrustedProxies: proxies}), nil, RoleChainListener, HTTP, "only to the http sink"},
{"trusted proxies on tcp", listener(config.AuthOptions{TrustedProxies: proxies}), nil, RoleListener, TCP, "only to the http sink"},
{"trusted proxies with certificate binding", listener(config.AuthOptions{TrustedProxies: proxies, Identity: "cn"}), f.serverTLS, RoleListener, HTTP, "TLS-terminating proxy"},
{"malformed trusted proxy", listener(config.AuthOptions{TrustedProxies: []string{"localhost"}}), nil, RoleListener, HTTP, "neither an address"},
{"trusted proxies on a dialer", dialer(config.AuthOptions{TrustedProxies: proxies}), &tls.Config{}, RoleDialer, HTTP, "only to listeners"},
{"unknown type", &config.AuthOptions{Type: "kerberos"}, mtlsListenerTLS(), config.Listener, TCP, `got "kerberos"`},
{"allow list without a type", &config.AuthOptions{Allow: []string{"edge-01"}}, mtlsListenerTLS(), config.Listener, TCP, "allow is set"},
{"credentials without a type", &config.AuthOptions{CredentialsFile: f.creds}, mtlsListenerTLS(), config.Listener, TCP, "credentials_file is set"},
{"no tls", &config.AuthOptions{Type: MethodMTLS}, nil, config.Listener, TCP, "requires tls.enabled"},
{"no client_auth", &config.AuthOptions{Type: MethodMTLS}, &tls.Config{}, config.Listener, TCP, "requires tls.client_auth"},
{"unknown identity", &config.AuthOptions{Type: MethodMTLS, Identity: "serial"}, mtlsListenerTLS(), config.Listener, TCP, `got "serial"`},
{"bad pattern", &config.AuthOptions{Type: MethodMTLS, AllowPatterns: []string{"^edge-("}}, mtlsListenerTLS(), config.Listener, TCP, "allow_patterns[0]"},
{"unknown binding", &config.AuthOptions{Type: MethodMTLS, NodeBinding: "maybe"}, mtlsListenerTLS(), config.ChainListener, TCP, `got "maybe"`},
{"binding on plain listener", &config.AuthOptions{Type: MethodMTLS, NodeBinding: BindingForce}, mtlsListenerTLS(), config.Listener, TCP, "only to chain sources"},
{"binding on dialer", &config.AuthOptions{Type: MethodMTLS, NodeBinding: BindingForce}, &tls.Config{}, config.Dialer, TCP, "only to chain sources"},
{"dialer skips verify", &config.AuthOptions{Type: MethodMTLS}, &tls.Config{InsecureSkipVerify: true}, config.Dialer, TCP, "insecure_skip_verify"},
{"mtls with a password", &config.AuthOptions{Type: MethodMTLS, PasswordFile: pw}, &tls.Config{}, config.Dialer, TCP, "password_file applies only"},
{"scram without tls", listener(config.AuthOptions{}), nil, config.Listener, TCP, "requires tls.enabled"},
{"scram without credentials", scram(config.AuthOptions{}), f.serverTLS, config.Listener, TCP, "requires credentials_file"},
{"scram with an allow list", listener(config.AuthOptions{Allow: []string{"edge-01"}}), f.serverTLS, config.Listener, TCP, "credentials file is the allow list"},
{"scram listener with a password", listener(config.AuthOptions{Username: "edge-01"}), f.serverTLS, config.Listener, TCP, "only to dialers"},
{"scram certificate binding without client_auth", listener(config.AuthOptions{Identity: "cn"}), f.serverTLS, config.Listener, TCP, "requires tls.client_auth"},
{"token lifetime on tcp", listener(config.AuthOptions{TokenLifetimeMS: 60000}), f.serverTLS, config.Listener, TCP, "only to HTTP listeners"},
{"token lifetime too short", listener(config.AuthOptions{TokenLifetimeMS: 900}), f.serverTLS, config.Listener, HTTP, "must be from 10000 to 86400000"},
{"token lifetime too long", listener(config.AuthOptions{TokenLifetimeMS: 1e13}), f.serverTLS, config.Listener, HTTP, "must be from 10000 to 86400000"},
{"scram listener without a certificate", listener(config.AuthOptions{}), &tls.Config{}, config.Listener, TCP, "listener certificate"},
{"scram dialer without a password", scram(config.AuthOptions{Username: "edge-01"}), &tls.Config{}, config.Dialer, TCP, "requires username and password_file"},
{"scram dialer with credentials", dialer(config.AuthOptions{CredentialsFile: f.creds}), &tls.Config{}, config.Dialer, TCP, "only to listeners"},
{"scram dialer skips verify", dialer(config.AuthOptions{}), &tls.Config{InsecureSkipVerify: true}, config.Dialer, TCP, "insecure_skip_verify"},
{"trusted proxies without a type", &config.AuthOptions{TrustedProxies: proxies}, nil, config.Listener, HTTP, "trusted_proxies is set"},
{"trusted proxies under mtls", &config.AuthOptions{Type: MethodMTLS, TrustedProxies: proxies}, mtlsListenerTLS(), config.Listener, HTTP, "trusted_proxies applies only"},
{"trusted proxies on a chain source", listener(config.AuthOptions{TrustedProxies: proxies}), nil, config.ChainListener, HTTP, "only to the http sink"},
{"trusted proxies on tcp", listener(config.AuthOptions{TrustedProxies: proxies}), nil, config.Listener, TCP, "only to the http sink"},
{"trusted proxies with certificate binding", listener(config.AuthOptions{TrustedProxies: proxies, Identity: "cn"}), f.serverTLS, config.Listener, HTTP, "TLS-terminating proxy"},
{"malformed trusted proxy", listener(config.AuthOptions{TrustedProxies: []string{"localhost"}}), nil, config.Listener, HTTP, "neither an address"},
{"trusted proxies on a dialer", dialer(config.AuthOptions{TrustedProxies: proxies}), &tls.Config{}, config.Dialer, HTTP, "only to listeners"},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
@@ -171,16 +171,16 @@ func TestNewValidation(t *testing.T) {
pinned := &tls.Config{InsecureSkipVerify: true, VerifyPeerCertificate: func([][]byte, [][]*x509.Certificate) error { return nil }}
for _, o := range []*config.AuthOptions{{Type: MethodMTLS}, dialer(config.AuthOptions{})} {
for _, c := range []*tls.Config{{}, pinned} {
if _, err := New(o, c, nil, RoleDialer, TCP); err != nil {
if _, err := New(o, c, nil, config.Dialer, TCP); err != nil {
t.Fatalf("dialer policy %s rejected: %v", o.Type, err)
}
}
}
if _, err := New(listener(config.AuthOptions{TokenLifetimeMS: 60000}), f.serverTLS, nil, RoleListener, HTTP); err != nil {
if _, err := New(listener(config.AuthOptions{TokenLifetimeMS: 60000}), f.serverTLS, nil, config.Listener, HTTP); err != nil {
t.Fatalf("scram listener rejected: %v", err)
}
// Behind proxies TLS ends at the proxy, so the hop may be plaintext
if _, err := New(listener(config.AuthOptions{TrustedProxies: proxies}), nil, nil, RoleListener, HTTP); err != nil {
if _, err := New(listener(config.AuthOptions{TrustedProxies: proxies}), nil, nil, config.Listener, HTTP); err != nil {
t.Fatalf("proxy-mode listener rejected: %v", err)
}
}
@@ -190,7 +190,7 @@ func TestAuthorizeMatching(t *testing.T) {
Type: MethodMTLS,
Allow: []string{"edge-01", " edge-02 "},
AllowPatterns: []string{`^relay-\d{2}$`},
}, mtlsListenerTLS(), nil, RoleListener, TCP)
}, mtlsListenerTLS(), nil, config.Listener, TCP)
if err != nil {
t.Fatalf("New: %v", err)
}
@@ -232,7 +232,7 @@ func TestAuthorizeMatching(t *testing.T) {
// Empty allow and allow_patterns admits any CA-vouched identity, but still
// records it and still refuses a certificate with no usable identity field
func TestAuthorizeUnrestricted(t *testing.T) {
p, err := New(&config.AuthOptions{Type: MethodMTLS}, mtlsListenerTLS(), nil, RoleListener, TCP)
p, err := New(&config.AuthOptions{Type: MethodMTLS}, mtlsListenerTLS(), nil, config.Listener, TCP)
if err != nil {
t.Fatalf("New: %v", err)
}
@@ -256,7 +256,7 @@ func TestAuthorizeUnrestricted(t *testing.T) {
func TestResolveNodeBindings(t *testing.T) {
newChain := func(binding string) *Policy {
p, err := New(&config.AuthOptions{Type: MethodMTLS, NodeBinding: binding}, mtlsListenerTLS(), nil, RoleChainListener, TCP)
p, err := New(&config.AuthOptions{Type: MethodMTLS, NodeBinding: binding}, mtlsListenerTLS(), nil, config.ChainListener, TCP)
if err != nil {
t.Fatalf("New(%q): %v", binding, err)
}
@@ -332,7 +332,7 @@ func TestIdentityApply(t *testing.T) {
func TestVerifyConnectionPinsServer(t *testing.T) {
p, err := New(&config.AuthOptions{Type: MethodMTLS, Allow: []string{"relay.internal"}},
&tls.Config{}, nil, RoleDialer, TCP)
&tls.Config{}, nil, config.Dialer, TCP)
if err != nil {
t.Fatalf("New: %v", err)
}
internal/authz/proxy.go
+2 -1
View File
@@ -11,6 +11,7 @@ import (
"sync"
"time"
"github.com/lixenwraith/logwisp/internal/config"
"github.com/lixenwraith/logwisp/internal/netacl"
"github.com/lixenwraith/auth"
@@ -34,7 +35,7 @@ type proxyMode struct {
func parseProxies(entries []string) (*proxyMode, error) {
m := &proxyMode{}
for _, e := range entries {
prefix, err := netacl.Parse(e, "tcp")
prefix, err := config.ParsePrefix(e, "tcp")
if err != nil {
return nil, fmt.Errorf("auth: trusted_proxies entry %q: %w", e, err)
}
internal/authz/proxy_test.go
+5 -5
View File
@@ -30,7 +30,7 @@ func (f forwarding) RoundTrip(r *http.Request) (*http.Response, error) {
// proxyListener is a plaintext http sink policy behind a proxy on loopback
func (f *fixture) proxyListener(t *testing.T) (*Policy, *httptest.Server) {
t.Helper()
l := f.listener(t, config.AuthOptions{TrustedProxies: []string{"127.0.0.1"}}, nil, RoleListener, HTTP)
l := f.listener(t, config.AuthOptions{TrustedProxies: []string{"127.0.0.1"}}, nil, config.Listener, HTTP)
mux := http.NewServeMux()
mux.HandleFunc("POST "+chain.AuthPath, func(w http.ResponseWriter, r *http.Request) { l.ServeAuth(w, r) })
mux.HandleFunc("GET /protected", func(w http.ResponseWriter, r *http.Request) {
@@ -49,7 +49,7 @@ func (f *fixture) proxyListener(t *testing.T) (*Policy, *httptest.Server) {
// when every hop is inside a proxy range.
func TestProxyModeTrustsOnlyItsProxies(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{TrustedProxies: []string{"127.0.0.1", "10.0.0.0/8"}}, nil, RoleListener, HTTP)
l := f.listener(t, config.AuthOptions{TrustedProxies: []string{"127.0.0.1", "10.0.0.0/8"}}, nil, config.Listener, HTTP)
for _, tc := range []struct{ name, peer, xff, proto, want string }{
{"untrusted peer", "192.0.2.1:5000", "203.0.113.7", "https", ""},
{"no proto", "127.0.0.1:5000", "203.0.113.7", "", ""},
@@ -91,7 +91,7 @@ func TestForwardedClientIsRefusedByTheRules(t *testing.T) {
t.Fatal(err)
}
l, err := New(&config.AuthOptions{Type: MethodSCRAM, CredentialsFile: f.creds, TrustedProxies: []string{"127.0.0.1"}},
nil, acl, RoleListener, HTTP)
nil, acl, config.Listener, HTTP)
if err != nil {
t.Fatal(err)
}
@@ -195,7 +195,7 @@ func TestBrowserSessionBehindProxy(t *testing.T) {
t.Fatalf("logins = %d", l.allowed.Load())
}
direct := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, HTTP)
direct := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, HTTP)
ds, _ := f.httpListener(t, direct, f.serverTLS)
client = f.httpClient(f.dialer(t, "edge-01", "edge-01-secret"))
proof := &auth.ClientFinalRequest{FullNonce: strings.Repeat("n", 32), ClientProof: "AAAA"}
@@ -210,7 +210,7 @@ func TestBrowserSessionBehindProxy(t *testing.T) {
// only unbound proofs (lw auth token --unbound) log in.
func TestOnlyUnboundLoginsBehindProxy(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{TrustedProxies: []string{"127.0.0.1"}}, f.serverTLS, RoleListener, HTTP)
l := f.listener(t, config.AuthOptions{TrustedProxies: []string{"127.0.0.1"}}, f.serverTLS, config.Listener, HTTP)
srv, _ := f.httpListener(t, l, f.serverTLS)
for _, unbound := range []bool{false, true} {
d := f.dialer(t, "edge-01", "edge-01-secret")
internal/authz/scram.go
+4 -42
View File
@@ -34,13 +34,8 @@ const (
ExchangeTimeout = 10 * time.Second
// DefaultTokenLifetime is how long a bearer token from /auth stays valid
DefaultTokenLifetime = 15 * time.Minute
// Token lifetime bounds: exp has whole seconds, a sub-second expires_in
// reads as unknown, and 10 s leaves room for renewal ahead of expiry
MinTokenLifetime = 10 * time.Second
MaxTokenLifetime = 24 * time.Hour
maxAuthLine = 4096 // pre-auth lines and /auth bodies come from unauthenticated peers
streamBuffer = 64 * 1024
maxAuthLine = 4096 // pre-auth lines and /auth bodies come from unauthenticated peers
streamBuffer = 64 * 1024
limitBurst = 10 // failed or abandoned exchanges per address before throttling
limitRate = 1 // per second
@@ -200,36 +195,11 @@ type scramListener struct {
}
func (p *Policy) compileSCRAM(o *config.AuthOptions, tlsCfg *tls.Config) error {
if len(o.Allow) > 0 || len(o.AllowPatterns) > 0 {
return fmt.Errorf("auth: allow and allow_patterns apply only to type %q; the credentials file is the allow list", MethodMTLS)
}
if p.role == RoleDialer {
if p.side == config.Dialer {
return p.compileSCRAMDialer(o)
}
switch {
case o.Username != "" || o.PasswordFile != "":
return errors.New("auth: username and password_file apply only to dialers")
case o.CredentialsFile == "":
return fmt.Errorf("auth: type %q requires credentials_file", MethodSCRAM)
case o.TokenLifetimeMS != 0 && (o.TokenLifetimeMS < MinTokenLifetime.Milliseconds() || o.TokenLifetimeMS > MaxTokenLifetime.Milliseconds()):
return fmt.Errorf("auth: token_lifetime_ms %d is outside %d (%s) to %d (%s)", o.TokenLifetimeMS,
MinTokenLifetime.Milliseconds(), MinTokenLifetime, MaxTokenLifetime.Milliseconds(), MaxTokenLifetime)
case o.TokenLifetimeMS > 0 && p.transport != HTTP:
return errors.New("auth: token_lifetime_ms applies only to HTTP listeners")
case len(o.TrustedProxies) > 0 && (p.role != RoleListener || p.transport != HTTP):
return errors.New("auth: trusted_proxies applies only to the http sink")
case len(o.TrustedProxies) > 0 && o.Identity != "":
return errors.New("auth: identity binds a client certificate, which a TLS-terminating proxy does not pass on; drop it or trusted_proxies")
}
if o.Identity != "" {
// Binds the certificate to the user: a peer needs its own of both
if tlsCfg.ClientAuth != tls.RequireAndVerifyClientCert {
return fmt.Errorf("auth: identity under type %q binds the client certificate to the user and requires tls.client_auth", MethodSCRAM)
}
var err error
if p.identity, err = identityMode(o.Identity); err != nil {
return err
}
p.identity = o.Identity // binds the certificate to the user
}
l := &scramListener{limit: netacl.NewTable(limitBurst, limitRate, limitPending)}
var err error
@@ -573,14 +543,6 @@ type scramDialer struct {
}
func (p *Policy) compileSCRAMDialer(o *config.AuthOptions) error {
switch {
case o.CredentialsFile != "", o.TokenLifetimeMS != 0, len(o.TrustedProxies) > 0:
return errors.New("auth: credentials_file, token_lifetime_ms and trusted_proxies apply only to listeners")
case o.Identity != "":
return fmt.Errorf("auth: identity on a dialer pins the server and applies only to type %q", MethodMTLS)
case o.Username == "" || o.PasswordFile == "":
return fmt.Errorf("auth: type %q on a dialer requires username and password_file", MethodSCRAM)
}
password, err := ReadPassword(o.PasswordFile)
if err != nil {
return err
internal/authz/scram_test.go
+24 -24
View File
@@ -99,7 +99,7 @@ func (f *fixture) pool(t *testing.T) *x509.CertPool {
return pool
}
func (f *fixture) listener(t *testing.T, o config.AuthOptions, tlsCfg *tls.Config, role Role, transport Transport) *Policy {
func (f *fixture) listener(t *testing.T, o config.AuthOptions, tlsCfg *tls.Config, role config.Side, transport Transport) *Policy {
t.Helper()
o.Type, o.CredentialsFile = MethodSCRAM, f.creds
p, err := New(&o, tlsCfg, nil, role, transport)
@@ -116,7 +116,7 @@ func (f *fixture) listener(t *testing.T, o config.AuthOptions, tlsCfg *tls.Confi
func (f *fixture) dialer(t *testing.T, user, password string) *Policy {
t.Helper()
pw := f.write(t, user+"-"+strings.ReplaceAll(password, "/", "_")+".pass", password+"\n")
p, err := New(&config.AuthOptions{Type: MethodSCRAM, Username: user, PasswordFile: pw}, f.clientTLS, nil, RoleDialer, TCP)
p, err := New(&config.AuthOptions{Type: MethodSCRAM, Username: user, PasswordFile: pw}, f.clientTLS, nil, config.Dialer, TCP)
if err != nil {
t.Fatal(err)
}
@@ -167,7 +167,7 @@ func (f *fixture) connect(t *testing.T, listener *Policy, serverTLS *tls.Config,
// as the final message reach the dialer through the reader Greet returns.
func TestTCPExchangeAdmitsAndKeepsTheStream(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleChainListener, TCP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.ChainListener, TCP)
res, r, err := f.connect(t, l, f.serverTLS, f.dialer(t, "edge-01", "edge-01-secret"), func(a *Admission) {
a.final = append(a.final, "first entry\n"...) // as TCP would coalesce them
if err := a.Accept(); err != nil {
@@ -190,7 +190,7 @@ func TestTCPExchangeAdmitsAndKeepsTheStream(t *testing.T) {
// probe cannot tell which accounts exist.
func TestTCPExchangeRefusesWrongPasswordAndUnknownUserAlike(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, TCP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, TCP)
var seen []string
for _, d := range []*Policy{f.dialer(t, "edge-01", "not-the-secret"), f.dialer(t, "edge-99", "edge-01-secret")} {
res, _, err := f.connect(t, l, f.serverTLS, d, nil)
@@ -211,7 +211,7 @@ func TestTCPExchangeRefusesAnotherCertificate(t *testing.T) {
f := newFixture(t)
other := f.serverTLS.Clone()
other.Certificates = []tls.Certificate{f.serverLeaf(t, "relay-b")}
l := f.listener(t, config.AuthOptions{}, other, RoleListener, TCP) // binds to relay-b
l := f.listener(t, config.AuthOptions{}, other, config.Listener, TCP) // binds to relay-b
res, _, err := f.connect(t, l, f.serverTLS, f.dialer(t, "edge-01", "edge-01-secret"), nil)
if !errors.Is(err, ErrRefused) || !errors.Is(res.err, ErrRefused) {
t.Fatalf("dialer %v, listener %v; want both refused", err, res.err)
@@ -225,7 +225,7 @@ func TestTCPExchangeRefusesAnotherCertificate(t *testing.T) {
// without scram answers credentials instead of leaving the dialer waiting.
func TestHelloAndPolicyMustAgree(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleChainListener, TCP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.ChainListener, TCP)
if res, _, _ := f.connect(t, l, f.serverTLS, nil, nil); !errors.Is(res.err, ErrRefused) {
t.Fatalf("scram listener admitted a plain hello: %v", res.err)
}
@@ -264,7 +264,7 @@ func TestGreetEndsWithContext(t *testing.T) {
// and the limiter at once instead of holding a slot until it times out.
func TestAbandonedExchangeReleasesItsSlot(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, TCP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, TCP)
sc, cc := net.Pipe()
server, client := tls.Server(sc, f.serverTLS), tls.Client(cc, f.clientTLS)
t.Cleanup(func() { sc.Close(); cc.Close() })
@@ -307,7 +307,7 @@ func TestAbandonedExchangeReleasesItsSlot(t *testing.T) {
// a reason and never as a final message it could take for admission.
func TestRejectAfterExchangeSendsReason(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleChainListener, TCP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.ChainListener, TCP)
_, _, err := f.connect(t, l, f.serverTLS, f.dialer(t, "edge-01", "edge-01-secret"), func(a *Admission) {
a.Reject("node label rejected")
})
@@ -323,7 +323,7 @@ func TestCertificateBindsToUser(t *testing.T) {
f := newFixture(t)
mtls := f.serverTLS.Clone()
mtls.ClientAuth = tls.RequireAndVerifyClientCert
l := f.listener(t, config.AuthOptions{Identity: "cn"}, mtls, RoleListener, TCP)
l := f.listener(t, config.AuthOptions{Identity: "cn"}, mtls, config.Listener, TCP)
if res, _, err := f.connect(t, l, mtls, f.dialer(t, "edge-01", "edge-01-secret"), nil); err != nil || res.err != nil {
t.Fatalf("own certificate and password: dialer %v, listener %v", err, res.err)
}
@@ -331,7 +331,7 @@ func TestCertificateBindsToUser(t *testing.T) {
t.Fatalf("edge-01's certificate admitted user edge-02: %v", res.err)
}
h := f.listener(t, config.AuthOptions{Identity: "cn"}, mtls, RoleListener, HTTP)
h := f.listener(t, config.AuthOptions{Identity: "cn"}, mtls, config.Listener, HTTP)
srv, _ := f.httpListener(t, h, mtls)
d := f.dialer(t, "edge-01", "edge-01-secret")
token, err := d.Token(t.Context(), f.httpClient(d), srv.URL)
@@ -354,7 +354,7 @@ func TestUnknownUserSaltSurvivesRestart(t *testing.T) {
f := newFixture(t)
var salts []string
for range 2 {
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, TCP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, TCP)
c, err := l.listener.server.Load().ProcessClientFirstMessage("nobody", strings.Repeat("n", 32))
if err != nil {
t.Fatal(err)
@@ -370,7 +370,7 @@ func TestUnknownUserSaltSurvivesRestart(t *testing.T) {
// refused before it is parsed.
func TestPreAuthInputIsCapped(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, TCP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, TCP)
sc, cc := net.Pipe()
server, client := tls.Server(sc, f.serverTLS), tls.Client(cc, f.clientTLS)
t.Cleanup(func() { sc.Close(); cc.Close() })
@@ -389,7 +389,7 @@ func TestPreAuthInputIsCapped(t *testing.T) {
t.Fatalf("oversized hello: %v", err)
}
h := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, HTTP)
h := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, HTTP)
srv, _ := f.httpListener(t, h, f.serverTLS)
hc := &http.Client{Transport: &http.Transport{TLSClientConfig: f.clientTLS}}
resp, err := hc.Post(srv.URL+chain.AuthPath, "application/json", bytes.NewReader(bytes.Repeat([]byte(" "), 2*maxAuthLine)))
@@ -406,7 +406,7 @@ func TestPreAuthInputIsCapped(t *testing.T) {
// proof, so a hostile server cannot obtain a cheaply guessable one.
func TestDialerRefusesACheapChallenge(t *testing.T) {
f := newFixture(t) // verifiers at t=1, m=64 KiB
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, TCP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, TCP)
d := f.dialer(t, "edge-01", "edge-01-secret")
minArgonTime, minArgonMemory = auth.DefaultArgonTime, auth.DefaultArgonMemory
t.Cleanup(func() { minArgonTime, minArgonMemory = 1, 64 })
@@ -422,7 +422,7 @@ func TestDialerRefusesACheapChallenge(t *testing.T) {
// answers "too many attempts" without running the exchange.
func TestOnlyFailedAttemptsAreThrottled(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, TCP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, TCP)
right := f.dialer(t, "edge-01", "edge-01-secret")
for i := range limitBurst + 1 {
if _, _, err := f.connect(t, l, f.serverTLS, right, nil); err != nil {
@@ -481,7 +481,7 @@ func get(t *testing.T, client *http.Client, url string, prepare func(*http.Reque
// header; a refused token can be dropped and replaced by a fresh login.
func TestHTTPTokenFlow(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, HTTP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, HTTP)
srv, _ := f.httpListener(t, l, f.serverTLS)
d := f.dialer(t, "edge-01", "edge-01-secret")
client := f.httpClient(d)
@@ -494,7 +494,7 @@ func TestHTTPTokenFlow(t *testing.T) {
if resp.StatusCode != http.StatusOK {
t.Fatalf("with a token: %d", resp.StatusCode)
}
other := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, HTTP)
other := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, HTTP)
foreignSrv, _ := f.httpListener(t, other, f.serverTLS)
foreign, err := f.dialer(t, "edge-01", "edge-01-secret").Token(t.Context(), f.httpClient(d), foreignSrv.URL)
if err != nil {
@@ -520,7 +520,7 @@ func TestHTTPTokenFlow(t *testing.T) {
// no token stored, whatever token it offers.
func TestForgedFinalYieldsNoToken(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, HTTP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, HTTP)
mux := http.NewServeMux()
mux.HandleFunc("POST "+chain.AuthPath, func(w http.ResponseWriter, r *http.Request) {
body := new(bytes.Buffer)
@@ -550,7 +550,7 @@ func TestForgedFinalYieldsNoToken(t *testing.T) {
// (through a proxy) included.
func TestPinnedCertificateAcrossConnections(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, HTTP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, HTTP)
a, _ := f.httpListener(t, l, f.serverTLS)
otherTLS := f.serverTLS.Clone()
otherTLS.Certificates = []tls.Certificate{f.serverLeaf(t, "relay-b")}
@@ -589,7 +589,7 @@ func TestPinnedCertificateAcrossConnections(t *testing.T) {
// its pending slot: otherwise four of them would lock the address out.
func TestRefusedHellosFreeTheirSlot(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, HTTP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, HTTP)
srv, _ := f.httpListener(t, l, f.serverTLS)
client := &http.Client{Transport: &http.Transport{TLSClientConfig: f.clientTLS}}
hello := func(user string) int {
@@ -660,7 +660,7 @@ func TestCredentialsFileValidation(t *testing.T) {
// exchange admits nobody rather than everybody.
func TestScramListenerAuthorizeFailsClosed(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, RoleListener, HTTP)
l := f.listener(t, config.AuthOptions{}, f.serverTLS, config.Listener, HTTP)
if _, err := l.Authorize(&tls.ConnectionState{}); !errors.Is(err, ErrRefused) {
t.Fatalf("Authorize under scram = %v, want a refusal", err)
}
@@ -670,7 +670,7 @@ func TestScramListenerAuthorizeFailsClosed(t *testing.T) {
// refused request each lifetime.
func TestTokenIsRenewedBeforeExpiry(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{TokenLifetimeMS: MinTokenLifetime.Milliseconds()}, f.serverTLS, RoleListener, HTTP)
l := f.listener(t, config.AuthOptions{TokenLifetimeMS: 10000}, f.serverTLS, config.Listener, HTTP)
srv, _ := f.httpListener(t, l, f.serverTLS)
d := f.dialer(t, "edge-01", "edge-01-secret")
client := f.httpClient(d)
@@ -686,7 +686,7 @@ func TestTokenIsRenewedBeforeExpiry(t *testing.T) {
t.Fatalf("logins within the lifetime = %d, want 1", n)
}
ahead := time.Until(time.Unix(0, d.dialer.renewAt.Load()))
if ahead <= 0 || ahead > MinTokenLifetime-5*time.Second {
if ahead <= 0 || ahead > 5*time.Second {
t.Fatalf("renewal scheduled %v ahead, want within lifetime - 5 s", ahead)
}
d.dialer.renewAt.Store(time.Now().UnixNano()) // the renewal point arrives
@@ -701,7 +701,7 @@ func TestTokenIsRenewedBeforeExpiry(t *testing.T) {
// any other refusal, or the token's expiry, ends it.
func TestPutOffRenewalKeepsTheValidToken(t *testing.T) {
f := newFixture(t)
l := f.listener(t, config.AuthOptions{TokenLifetimeMS: MinTokenLifetime.Milliseconds()}, f.serverTLS, RoleListener, HTTP)
l := f.listener(t, config.AuthOptions{TokenLifetimeMS: 10000}, f.serverTLS, config.Listener, HTTP)
srv, _ := f.httpListener(t, l, f.serverTLS)
// An impostor with another certificate of the same CA
var answer, stolen atomic.Int64
internal/config/config.go
+143 -190
View File
@@ -6,9 +6,9 @@ package config
type Config struct {
Quiet bool `toml:"quiet"`
// Console sinks without their own color: "auto" (a terminal, NO_COLOR
// unset, TERM not dumb), "always", "never"
Color string `toml:"color"`
// Console sinks without their own color: "auto" is a terminal with
// NO_COLOR unset and TERM not dumb
Color string `toml:"color" lw:"enum=auto|always|never"`
// Runtime behavior flags
StatusReporter bool `toml:"status_reporter"`
@@ -26,17 +26,10 @@ type Config struct {
// LogConfig represents the logging configuration for the LogWisp application itself
type LogConfig struct {
// Output mode: "file", "stdout", "stderr", "split", "all", "none"
Output string `toml:"output"`
// Log level: "debug", "info", "warn", "error"
Level string `toml:"level"`
// Format: "raw", "txt", "json"
Format string `toml:"format"`
// Sanitization policy for console output
Sanitization string `toml:"sanitization"`
Output string `toml:"output" lw:"enum=file|stdout|stderr|split|all|none"`
Level string `toml:"level" lw:"enum=debug|info|warn|error"`
Format string `toml:"format" lw:"enum=raw|txt|json,zero=the log default"`
Sanitization string `toml:"sanitization" lw:"enum=raw|json|txt|shell,zero=the log default"` // console output
// File output settings (when Output includes "file" or "all")
File *LogFileConfig `toml:"file"`
@@ -65,8 +58,7 @@ type LogFileConfig struct {
// LogConsoleConfig defines settings for console-based application logging
type LogConsoleConfig struct {
// Target for console output: "stdout", "stderr"
Target string `toml:"target"`
Target string `toml:"target" lw:"enum=stdout|stderr|split"`
}
// --- Pipeline ---
@@ -90,29 +82,25 @@ type FlowConfig struct {
Format *FormatConfig `toml:"format"`
}
// --- Heartbeat Options ---
// --- Flow stages ---
// HeartbeatConfig defines settings for periodic keep-alive or status messages
type HeartbeatConfig struct {
Enabled bool `toml:"enabled"`
IntervalMS int64 `toml:"interval_ms"`
IncludeTimestamp bool `toml:"include_timestamp"`
IncludeStats bool `toml:"include_stats"`
Format string `toml:"format"`
Enabled bool `toml:"enabled" help:"send heartbeats"`
IntervalMS int64 `toml:"interval_ms" default:"1000" lw:"min=100" help:"milliseconds between heartbeats"`
IncludeTimestamp bool `toml:"include_timestamp" help:"stamp each heartbeat with its time"`
IncludeStats bool `toml:"include_stats" help:"add the pipeline's counters"`
Format string `toml:"format" default:"txt" lw:"enum=txt|json|raw" help:"heartbeat format"`
}
// --- Formatter Options ---
// FormatConfig is a polymorphic struct representing log entry formatting options
// FormatConfig is how a pipeline writes its entries
type FormatConfig struct {
Type string `toml:"type"` // "json", "txt", "raw"
Flags int64 `toml:"flags,omitempty"`
TimestampFormat string `toml:"timestamp_format,omitempty"`
SanitizerPolicy string `toml:"sanitizer_policy,omitempty"` // "raw", "json", "txt", "shell"
Type string `toml:"type" default:"raw" lw:"enum=raw|txt|text|json" help:"entry format; text is txt"`
Flags int64 `toml:"flags,omitempty" lw:"zero=by type" help:"formatter flags"`
TimestampFormat string `toml:"timestamp_format,omitempty" lw:"zero=RFC 3339" help:"Go time layout of timestamps"`
SanitizerPolicy string `toml:"sanitizer_policy,omitempty" lw:"enum=raw|json|txt|shell,zero=by type" help:"how control characters are escaped"`
}
// --- Rate Limit Options ---
// RateLimitPolicy defines the action to take when a rate limit is exceeded
type RateLimitPolicy int
@@ -125,26 +113,20 @@ const (
// RateLimitConfig defines the configuration for pipeline-level rate limiting
type RateLimitConfig struct {
// Rate is the number of log entries allowed per second. Default: 0 (disabled)
Rate float64 `toml:"rate"`
// Burst is the maximum number of log entries that can be sent in a short burst. Defaults to the Rate
Burst float64 `toml:"burst"`
// Policy defines the action to take when the limit is exceeded. "pass" or "drop"
Policy string `toml:"policy"`
// MaxEntrySizeBytes is the maximum allowed size for a single log entry. 0 = no limit
MaxEntrySizeBytes int64 `toml:"max_entry_size_bytes"`
Rate float64 `toml:"rate" lw:"min=0,zero=off" help:"entries a second"`
Burst float64 `toml:"burst" lw:"min=0,zero=the rate" help:"entries at once"`
Policy string `toml:"policy" default:"pass" lw:"enum=pass|drop" help:"drop entries over the rate, or pass them"`
MaxEntrySizeBytes int64 `toml:"max_entry_size_bytes" lw:"min=0,zero=unlimited" help:"drop larger entries"`
}
// --- Filter Options ---
// FilterType represents the filter's behavior (include or exclude)
type FilterType string
const (
// FilterTypeInclude specifies that only matching logs will pass
FilterTypeInclude FilterType = "include" // Whitelist - only matching logs pass
FilterTypeInclude FilterType = "include"
// FilterTypeExclude specifies that matching logs will be dropped
FilterTypeExclude FilterType = "exclude" // Blacklist - matching logs are dropped
FilterTypeExclude FilterType = "exclude"
)
// FilterLogic represents how multiple filter patterns are combined
@@ -152,19 +134,19 @@ type FilterLogic string
const (
// FilterLogicOr specifies that a match on any pattern is sufficient
FilterLogicOr FilterLogic = "or" // Match any pattern
FilterLogicOr FilterLogic = "or"
// FilterLogicAnd specifies that all patterns must match
FilterLogicAnd FilterLogic = "and" // Match all patterns
FilterLogicAnd FilterLogic = "and"
)
// FilterConfig represents the configuration for a single filter
type FilterConfig struct {
Type FilterType `toml:"type"`
Logic FilterLogic `toml:"logic"`
Patterns []string `toml:"patterns"`
Type FilterType `toml:"type" default:"include" lw:"enum=include|exclude" help:"pass only matching entries, or drop them"`
Logic FilterLogic `toml:"logic" default:"or" lw:"enum=or|and" help:"match any pattern, or all"`
Patterns []string `toml:"patterns" lw:"hint=regex" help:"RE2 patterns"`
}
// --- Source Options ---
// --- Sources ---
// PluginSourceConfig represents a source plugin instance configuration
type PluginSourceConfig struct {
@@ -179,38 +161,38 @@ type NullSourceOptions struct{}
// RandomSourceOptions defines settings for a random log generator source
type RandomSourceOptions struct {
IntervalMS int64 `toml:"interval_ms"`
JitterMS int64 `toml:"jitter_ms"`
Format string `toml:"format"`
Length int64 `toml:"length"`
Special bool `toml:"special"`
IntervalMS int64 `toml:"interval_ms" default:"500" lw:"min=1" help:"milliseconds between entries"`
JitterMS int64 `toml:"jitter_ms" lw:"min=0,zero=none" help:"random milliseconds added to the interval, at most the interval"`
Format string `toml:"format" default:"txt" lw:"enum=raw|txt|json" help:"entry format"`
Length int64 `toml:"length" default:"20" lw:"min=1" help:"message characters"`
Special bool `toml:"special" help:"mix in control characters"`
}
// FileSourceOptions defines settings for a file-based source
type FileSourceOptions struct {
Directory string `toml:"directory"`
Pattern string `toml:"pattern"` // glob pattern
CheckIntervalMS int64 `toml:"check_interval_ms"`
Raw bool `toml:"raw"` // keep the whole line as the message, never parse it
From string `toml:"from"` // "end" (default) or "start" of a newly discovered file
Directory string `toml:"directory" lw:"required,hint=dir" help:"directory whose files are followed"`
Pattern string `toml:"pattern" default:"*" lw:"hint=glob" help:"glob the file names match"`
CheckIntervalMS int64 `toml:"check_interval_ms" default:"100" lw:"min=10" help:"milliseconds between scans for files"`
Raw bool `toml:"raw" help:"keep the whole line as the message, never parse it"`
From string `toml:"from" default:"end" lw:"enum=start|end" help:"where reading a newly found file starts"`
}
// ConsoleSourceOptions defines settings for a stdin-based source
type ConsoleSourceOptions struct {
BufferSize int64 `toml:"buffer_size"`
BufferSize int64 `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued"`
}
// TCPChainSourceOptions defines settings for a stdlib TCP listener ingesting
// NDJSON entries from upstream logwisp tcp_chain sinks
type TCPChainSourceOptions struct {
TLS *TLSOptions `toml:"tls"`
Host string `toml:"host"`
Port int64 `toml:"port"`
BufferSize int64 `toml:"buffer_size"`
MaxConnections int64 `toml:"max_connections"` // 0 = unlimited
ReadTimeoutMS int64 `toml:"read_timeout_ms"` // per-connection idle deadline, 0 = none
HelloTimeoutMS int64 `toml:"hello_timeout_ms"` // preamble deadline
TrustNode bool `toml:"trust_node"` // false: force node label from remote address
Host string `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on; :: for IPv6"`
Port int64 `toml:"port" lw:"required,min=1,max=65535" help:"port to listen on"`
BufferSize int64 `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued"`
MaxConnections int64 `toml:"max_connections" lw:"min=0,zero=unlimited" help:"connections at once"`
ReadTimeoutMS int64 `toml:"read_timeout_ms" lw:"min=0,zero=none" help:"milliseconds a connection may idle"`
HelloTimeoutMS int64 `toml:"hello_timeout_ms" default:"10000" lw:"min=1" help:"milliseconds for a peer's hello"`
TrustNode bool `toml:"trust_node" default:"true" help:"keep the node label a peer declares; off, the label is its address"`
Auth *AuthOptions `toml:"auth"`
ACL *ACLOptions `toml:"acl"`
}
@@ -219,18 +201,18 @@ type TCPChainSourceOptions struct {
// NDJSON batches from upstream logwisp http_chain sinks
type HTTPChainSourceOptions struct {
TLS *TLSOptions `toml:"tls"`
Host string `toml:"host"`
Port int64 `toml:"port"`
IngestPath string `toml:"ingest_path"`
BufferSize int64 `toml:"buffer_size"`
MaxBodyBytes int64 `toml:"max_body_bytes"` // per-request cap
ReadTimeoutMS int64 `toml:"read_timeout_ms"` // full request read deadline
TrustNode bool `toml:"trust_node"` // false: force node label from remote address
Host string `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on; :: for IPv6"`
Port int64 `toml:"port" lw:"required,min=1,max=65535" help:"port to listen on"`
IngestPath string `toml:"ingest_path" default:"/ingest" lw:"hint=path" help:"path batches are posted to"`
BufferSize int64 `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued"`
MaxBodyBytes int64 `toml:"max_body_bytes" default:"8388608" lw:"min=1" help:"largest batch accepted"`
ReadTimeoutMS int64 `toml:"read_timeout_ms" default:"30000" lw:"min=1" help:"milliseconds to read a request"`
TrustNode bool `toml:"trust_node" default:"true" help:"keep the node label a peer declares; off, the label is its address"`
Auth *AuthOptions `toml:"auth"`
ACL *ACLOptions `toml:"acl"`
}
// --- Sink Options ---
// --- Sinks ---
// PluginSinkConfig represents a sink plugin instance configuration
type PluginSinkConfig struct {
@@ -245,35 +227,35 @@ type NullSinkOptions struct{}
// ConsoleSinkOptions defines settings for a console-based sink
type ConsoleSinkOptions struct {
Target string `toml:"target"` // "stdout", "stderr"
BufferSize int64 `toml:"buffer_size"`
Escape string `toml:"escape"` // control characters as <hex>: "auto" (on a terminal), "always", "never"
Color string `toml:"color"` // level names in color: "auto", "always", "never"; default: the top-level color
Target string `toml:"target" default:"stdout" lw:"enum=stdout|stderr" help:"stream written to"`
BufferSize int64 `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued"`
Escape string `toml:"escape" default:"auto" lw:"enum=auto|always|never" help:"write control characters as <hex>; auto on a terminal"`
Color string `toml:"color" default:"auto" lw:"enum=auto|always|never" help:"level names in color; unset, the top-level color"`
}
// FileSinkOptions defines settings for a file-based sink
type FileSinkOptions struct {
Directory string `toml:"directory"`
Name string `toml:"name"`
MaxSizeMB int64 `toml:"max_size_mb"`
MaxTotalSizeMB int64 `toml:"max_total_size_mb"`
MinDiskFreeMB int64 `toml:"min_disk_free_mb"`
RetentionHours float64 `toml:"retention_hours"`
BufferSize int64 `toml:"buffer_size"`
FlushIntervalMs int64 `toml:"flush_interval_ms"`
Directory string `toml:"directory" lw:"required,hint=dir" help:"directory written to"`
Name string `toml:"name" lw:"required" help:"base name of the files"`
MaxSizeMB int64 `toml:"max_size_mb" default:"100" lw:"min=1" help:"size at which a file rotates"`
MaxTotalSizeMB int64 `toml:"max_total_size_mb" default:"1000" lw:"min=1" help:"size of all files, beyond which the oldest go"`
MinDiskFreeMB int64 `toml:"min_disk_free_mb" lw:"zero=none" help:"free space kept on the disk; negative means 100"`
RetentionHours float64 `toml:"retention_hours" default:"168" lw:"min=0" help:"hours a file is kept"`
BufferSize int64 `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued"`
FlushIntervalMs int64 `toml:"flush_interval_ms" default:"100" lw:"min=1" help:"milliseconds between flushes"`
}
// TCPSinkOptions defines settings for a TCP server sink
type TCPSinkOptions struct {
TLS *TLSOptions `toml:"tls"`
Host string `toml:"host"`
Port int64 `toml:"port"`
BufferSize int64 `toml:"buffer_size"` // sink input queue
ClientBufferSize int64 `toml:"client_buffer_size"` // per-client send queue
WriteTimeoutMS int64 `toml:"write_timeout_ms"` // per-write deadline
KeepAlive bool `toml:"keep_alive"`
KeepAlivePeriodMS int64 `toml:"keep_alive_period_ms"`
MaxConnections int64 `toml:"max_connections"` // 0 = unlimited
Host string `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on; :: for IPv6"`
Port int64 `toml:"port" lw:"required,min=1,max=65535" help:"port to listen on"`
BufferSize int64 `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued for all clients"`
ClientBufferSize int64 `toml:"client_buffer_size" default:"256" lw:"min=1" help:"entries queued per client"`
WriteTimeoutMS int64 `toml:"write_timeout_ms" default:"5000" lw:"min=1" help:"milliseconds a write may take before the client is dropped"`
KeepAlive bool `toml:"keep_alive" default:"true" help:"TCP keep-alive"`
KeepAlivePeriodMS int64 `toml:"keep_alive_period_ms" default:"30000" lw:"min=1" help:"milliseconds idle before a keep-alive probe"`
MaxConnections int64 `toml:"max_connections" lw:"min=0,zero=unlimited" help:"clients at once"`
Auth *AuthOptions `toml:"auth"`
ACL *ACLOptions `toml:"acl"`
}
@@ -281,17 +263,17 @@ type TCPSinkOptions struct {
// HTTPSinkOptions defines settings for an HTTP SSE server sink
type HTTPSinkOptions struct {
TLS *TLSOptions `toml:"tls"`
Host string `toml:"host"`
Port int64 `toml:"port"`
StreamPath string `toml:"stream_path"`
StatusPath string `toml:"status_path"`
BufferSize int64 `toml:"buffer_size"` // sink input queue
ClientBufferSize int64 `toml:"client_buffer_size"` // per-client send queue
WriteTimeoutMS int64 `toml:"write_timeout_ms"` // per-SSE-write deadline, 0 = none
MaxConnections int64 `toml:"max_connections"` // 0 = unlimited
ReplayLines int64 `toml:"replay_lines"` // entries a new stream gets first, 0 = none
LoginPage bool `toml:"login_page"` // GET /auth/login, needs auth.trusted_proxies
ViewerPage bool `toml:"viewer_page"` // GET /auth/view, needs login_page
Host string `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on; :: for IPv6"`
Port int64 `toml:"port" lw:"required,min=1,max=65535" help:"port to listen on"`
StreamPath string `toml:"stream_path" default:"/stream" lw:"hint=path" help:"path of the event stream"`
StatusPath string `toml:"status_path" default:"/status" lw:"hint=path" help:"path of the status document"`
BufferSize int64 `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued for all clients"`
ClientBufferSize int64 `toml:"client_buffer_size" default:"256" lw:"min=1" help:"entries queued per client"`
WriteTimeoutMS int64 `toml:"write_timeout_ms" lw:"min=0,zero=none" help:"milliseconds an event write may take"`
MaxConnections int64 `toml:"max_connections" lw:"min=0,zero=unlimited" help:"clients at once"`
ReplayLines int64 `toml:"replay_lines" lw:"min=0,zero=none" help:"recent entries a new stream gets first"`
LoginPage bool `toml:"login_page" help:"serve the login page; scram behind auth.trusted_proxies"`
ViewerPage bool `toml:"viewer_page" help:"serve the viewer to users logged in; needs login_page"`
Auth *AuthOptions `toml:"auth"`
ACL *ACLOptions `toml:"acl"`
}
@@ -300,16 +282,16 @@ type HTTPSinkOptions struct {
// entries to a downstream logwisp tcp_chain source
type TCPChainSinkOptions struct {
TLS *TLSOptions `toml:"tls"`
Node string `toml:"node"` // origin label, default: os.Hostname()
Host string `toml:"host"`
Port int64 `toml:"port"`
BufferSize int64 `toml:"buffer_size"`
DialTimeoutMS int64 `toml:"dial_timeout_ms"`
WriteTimeoutMS int64 `toml:"write_timeout_ms"`
BackoffMinMS int64 `toml:"backoff_min_ms"`
BackoffMaxMS int64 `toml:"backoff_max_ms"`
KeepAlivePeriodMS int64 `toml:"keep_alive_period_ms"`
KeepAlive bool `toml:"keep_alive"`
Node string `toml:"node" lw:"zero=the host name" help:"origin label of the entries"`
Host string `toml:"host" lw:"required,hint=host" help:"host of the tcp_chain source"`
Port int64 `toml:"port" lw:"required,min=1,max=65535" help:"port of the tcp_chain source"`
BufferSize int64 `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued"`
DialTimeoutMS int64 `toml:"dial_timeout_ms" default:"5000" lw:"min=1" help:"milliseconds to connect"`
WriteTimeoutMS int64 `toml:"write_timeout_ms" default:"5000" lw:"min=1" help:"milliseconds a write may take"`
BackoffMinMS int64 `toml:"backoff_min_ms" default:"500" lw:"min=1" help:"first wait before reconnecting"`
BackoffMaxMS int64 `toml:"backoff_max_ms" default:"30000" lw:"min=1" help:"longest wait before reconnecting; below the first, the default"`
KeepAlivePeriodMS int64 `toml:"keep_alive_period_ms" default:"30000" lw:"min=1" help:"milliseconds idle before a keep-alive probe"`
KeepAlive bool `toml:"keep_alive" default:"true" help:"TCP keep-alive"`
Auth *AuthOptions `toml:"auth"`
}
@@ -317,17 +299,17 @@ type TCPChainSinkOptions struct {
// NDJSON batches to a downstream logwisp http_chain source
type HTTPChainSinkOptions struct {
TLS *TLSOptions `toml:"tls"`
Node string `toml:"node"` // origin label, default: os.Hostname()
Host string `toml:"host"`
Port int64 `toml:"port"`
IngestPath string `toml:"ingest_path"`
BufferSize int64 `toml:"buffer_size"`
MaxBatchCount int64 `toml:"max_batch_count"`
MaxBatchBytes int64 `toml:"max_batch_bytes"`
FlushIntervalMS int64 `toml:"flush_interval_ms"`
RequestTimeoutMS int64 `toml:"request_timeout_ms"` // covers dial + write + response
BackoffMinMS int64 `toml:"backoff_min_ms"`
BackoffMaxMS int64 `toml:"backoff_max_ms"`
Node string `toml:"node" lw:"zero=the host name" help:"origin label of the entries"`
Host string `toml:"host" lw:"required,hint=host" help:"host of the http_chain source"`
Port int64 `toml:"port" lw:"required,min=1,max=65535" help:"port of the http_chain source"`
IngestPath string `toml:"ingest_path" default:"/ingest" lw:"hint=path" help:"path batches are posted to"`
BufferSize int64 `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued"`
MaxBatchCount int64 `toml:"max_batch_count" default:"100" lw:"min=1" help:"entries in a batch"`
MaxBatchBytes int64 `toml:"max_batch_bytes" default:"1048576" lw:"min=1" help:"bytes in a batch"`
FlushIntervalMS int64 `toml:"flush_interval_ms" default:"1000" lw:"min=1" help:"milliseconds before a partial batch is sent"`
RequestTimeoutMS int64 `toml:"request_timeout_ms" default:"10000" lw:"min=1" help:"milliseconds for a request: dial, write and response"`
BackoffMinMS int64 `toml:"backoff_min_ms" default:"500" lw:"min=1" help:"first wait before retrying"`
BackoffMaxMS int64 `toml:"backoff_max_ms" default:"30000" lw:"min=1" help:"longest wait before retrying; below the first, the default"`
Auth *AuthOptions `toml:"auth"`
}
@@ -337,50 +319,33 @@ type HTTPChainSinkOptions struct {
// beside `tls`: TLS answers "is this channel private", auth answers "may this
// peer do this". Listeners verify (mtls: certificate identity; scram: password
// via credentials_file); dialers prove themselves (scram) or pin the server
// (mtls). Validation is per role in internal/authz.
// (mtls). AuthOptions.Check holds the rules of each side.
type AuthOptions struct {
// Method: "none" (default) | "mtls" | "scram"
Type string `toml:"type"`
// Certificate field carrying the identity: "cn" (default) | "san_dns" |
// "san_uri" | "san_email". Under scram it binds the client certificate to
// the user: the field must equal the SCRAM username.
Identity string `toml:"identity"`
// mtls only: exact identities and RE2 patterns (anchor them yourself).
// Both empty admits any identity the CA vouches for.
Allow []string `toml:"allow"`
AllowPatterns []string `toml:"allow_patterns"`
// Chain sources only: "none" | "assert" | "force" (default "force").
// Overrides trust_node.
NodeBinding string `toml:"node_binding"`
// scram listeners: verifier file written by `lw auth add-user`, and
// the bearer token lifetime on HTTP listeners (default 15 minutes)
CredentialsFile string `toml:"credentials_file"`
TokenLifetimeMS int64 `toml:"token_lifetime_ms"`
// scram http sink: addresses or CIDRs of the reverse proxies that end the
// browsers' TLS; logins are then unbound and sessions may be cookies
TrustedProxies []string `toml:"trusted_proxies"`
// scram dialers: the identity presented and the file holding its password
Username string `toml:"username"`
PasswordFile string `toml:"password_file"`
Type string `toml:"type" default:"none" lw:"enum=none|mtls|scram" help:"how peers authenticate"`
Identity string `toml:"identity" lw:"enum=cn|san_dns|san_uri|san_email,zero=cn under mtls; no binding under scram" help:"certificate field naming the peer; under scram it must equal the user"`
Allow []string `toml:"allow" help:"mtls: identities admitted; with allow_patterns empty, any the CA issued"`
AllowPatterns []string `toml:"allow_patterns" lw:"hint=regex" help:"mtls: RE2 patterns of identities admitted; anchor them"`
NodeBinding string `toml:"node_binding" lw:"listener,enum=none|assert|force,zero=force" help:"chain sources: how a peer's identity binds its node label; overrides trust_node"`
CredentialsFile string `toml:"credentials_file" lw:"listener,hint=file" help:"scram: verifiers, from lw auth add-user"`
// exp has whole seconds, a sub-second expires_in reads as unknown, and 10 s
// leaves room for renewal ahead of expiry
TokenLifetimeMS int64 `toml:"token_lifetime_ms" lw:"listener,min=10000,max=86400000,zero=15 minutes" help:"scram on HTTP: bearer token lifetime"`
TrustedProxies []string `toml:"trusted_proxies" lw:"listener,hint=cidr" help:"scram on the http sink: proxies ending the browsers' TLS; logins are then unbound"`
Username string `toml:"username" lw:"dialer" help:"scram: the user this dialer logs in as"`
PasswordFile string `toml:"password_file" lw:"dialer,hint=file" help:"scram: file holding the user's password"`
}
// ACLOptions is a listener's address rules, entries being addresses or CIDRs:
// deny wins, then a set allow list admits only its entries. proxy_from lists
// the L4 proxies whose PROXY header names the client the rules then see; the
// per-client limits count that client. Validation is in internal/netacl.
// per-client limits count that client. ACLOptions.Check holds the rules.
type ACLOptions struct {
Allow []string `toml:"allow"`
Deny []string `toml:"deny"`
ProxyProtocol string `toml:"proxy_protocol"` // off, optional or required
ProxyFrom []string `toml:"proxy_from"`
MaxConnectionsPerClient int64 `toml:"max_connections_per_client"` // 0 = unlimited
RequestsPerSecondPerClient float64 `toml:"requests_per_second_per_client"` // HTTP listeners; 0 = unlimited
Allow []string `toml:"allow" lw:"hint=cidr" help:"addresses admitted; empty, all deny does not list"`
Deny []string `toml:"deny" lw:"hint=cidr" help:"addresses refused"`
ProxyProtocol string `toml:"proxy_protocol" default:"off" lw:"enum=off|optional|required" help:"read a PROXY header from proxy_from"`
ProxyFrom []string `toml:"proxy_from" lw:"hint=cidr" help:"proxies that may send a PROXY header"`
MaxConnectionsPerClient int64 `toml:"max_connections_per_client" lw:"min=0,zero=unlimited" help:"connections a client may hold"`
RequestsPerSecondPerClient float64 `toml:"requests_per_second_per_client" lw:"min=0,zero=unlimited" help:"HTTP: requests a client may make each second"`
}
// --- TLS Options ---
@@ -391,30 +356,18 @@ type ACLOptions struct {
// dialers verify the server with ca_file and server_name, or pin_sha256, and
// may present cert_file and key_file.
type TLSOptions struct {
Enabled bool `toml:"enabled"`
// Local identity: required for listeners, optional for dialers (mTLS)
CertFile string `toml:"cert_file"`
KeyFile string `toml:"key_file"`
// Listeners without cert_file: a certificate on a key made at startup,
// self-signed or signed by the issuer CA, for hosts plus the host option,
// os.Hostname(), localhost and the loopback addresses
SelfSigned bool `toml:"self_signed"`
IssuerCertFile string `toml:"issuer_cert_file"`
IssuerKeyFile string `toml:"issuer_key_file"`
Hosts []string `toml:"hosts"`
// Listener-side peer verification (mTLS)
ClientAuth bool `toml:"client_auth"`
ClientCAFile string `toml:"client_ca_file"`
// Dialer-side peer verification
CAFile string `toml:"ca_file"` // empty = system trust store
ServerName string `toml:"server_name"` // default: config host
InsecureSkipVerify bool `toml:"insecure_skip_verify"`
PinSHA256 string `toml:"pin_sha256"` // sha256//BASE64 of the server key, ';' between several
// Minimum protocol version: "1.2" | "1.3" (default "1.3")
MinVersion string `toml:"min_version"`
Enabled bool `toml:"enabled" help:"use TLS"`
CertFile string `toml:"cert_file" lw:"hint=file" help:"certificate presented; a dialer's for mTLS"`
KeyFile string `toml:"key_file" lw:"hint=file" help:"key of cert_file"`
SelfSigned bool `toml:"self_signed" lw:"listener" help:"a self-signed certificate on a key made at startup"`
IssuerCertFile string `toml:"issuer_cert_file" lw:"listener,hint=file" help:"CA certificate (lw tls ca) that issues the listener's at startup"`
IssuerKeyFile string `toml:"issuer_key_file" lw:"listener,hint=file" help:"key of issuer_cert_file"`
Hosts []string `toml:"hosts" lw:"listener" help:"names and addresses a made certificate carries beyond host, this machine's and loopback"`
ClientAuth bool `toml:"client_auth" lw:"listener" help:"require and verify client certificates"`
ClientCAFile string `toml:"client_ca_file" lw:"listener,hint=file" help:"CA bundle verifying client certificates"`
CAFile string `toml:"ca_file" lw:"dialer,hint=file,zero=the system roots" help:"CA bundle verifying the server"`
ServerName string `toml:"server_name" lw:"dialer,zero=the host" help:"name the server's certificate must carry"`
InsecureSkipVerify bool `toml:"insecure_skip_verify" lw:"dialer" help:"verify nothing of the server; never in production"`
PinSHA256 string `toml:"pin_sha256" lw:"dialer" help:"sha256//BASE64 of the server's key, ';' between several; replaces ca_file"`
MinVersion string `toml:"min_version" default:"1.3" lw:"enum=1.2|1.3" help:"lowest TLS version"`
}
internal/config/loader_test.go
+11 -18
View File
@@ -168,9 +168,8 @@ write_timeout_ms = 9007199254740993
[pipelines.plugin_sinks.config.tls]
enabled = false
server_name = "relay.example"
[pipelines.plugin_sinks.config.auth]
type = "none"
allow = ["one", "two"]
[pipelines.plugin_sinks.config.acl]
allow = ["10.0.0.1", "10.0.0.2"]
`)
m, err := Load(Args{File: "plugins.toml"})
if err != nil {
@@ -181,18 +180,18 @@ allow = ["one", "two"]
if err != nil {
t.Fatal(err)
}
opts := TCPSinkOptions{Host: "127.0.0.1", BufferSize: 1000}
if err := Scan(cfg.Pipelines[0].PluginSinks[0].Config, &opts); err != nil {
opts, err := Decode[TCPSinkOptions]("sink", "tcp", cfg.Pipelines[0].PluginSinks[0].Config)
if err != nil {
t.Fatal(err)
}
if opts.Host != "127.0.0.1" || opts.BufferSize != 1000 || opts.Port != 9000 || opts.WriteTimeoutMS != 9007199254740993 ||
opts.TLS == nil || opts.TLS.ServerName != "relay.example" || opts.Auth == nil || !reflect.DeepEqual(opts.Auth.Allow, []string{"one", "two"}) {
if opts.Host != "0.0.0.0" || opts.BufferSize != 1000 || !opts.KeepAlive || opts.Port != 9000 || opts.WriteTimeoutMS != 9007199254740993 ||
opts.TLS == nil || opts.TLS.ServerName != "relay.example" || opts.ACL == nil || !reflect.DeepEqual(opts.ACL.Allow, []string{"10.0.0.1", "10.0.0.2"}) {
t.Fatalf("plugin defaults, nested tables or integer precision lost: %+v", opts)
}
}
// A misspelled key, even inside a nested tls or auth table, fails plugin
// construction instead of leaving the protection it named switched off.
// A misspelled key, even inside a nested tls or auth table, fails the load
// instead of leaving the protection it named switched off.
func TestPluginConfigRejectsUnknownKeys(t *testing.T) {
isolateConfig(t)
for key, table := range map[string]string{
@@ -213,16 +212,10 @@ type = "tcp"
port = 9000
`+table+"\n")
m, err := Load(Args{File: "plugins.toml"})
if err != nil {
t.Fatal(err)
if err == nil {
m.Close()
}
cfg, err := m.Snapshot()
m.Close()
if err != nil {
t.Fatal(err)
}
err = Scan(cfg.Pipelines[0].PluginSinks[0].Config, &TCPSinkOptions{})
if err == nil || !strings.Contains(err.Error(), `"`+key+`"`) {
if err == nil || !strings.Contains(err.Error(), `pipelines[0].plugin_sinks[out].config: unknown key "`+key+`"`) {
t.Errorf("unknown key %s: err = %v", key, err)
}
}
internal/config/schema.go
+370
View File
@@ -0,0 +1,370 @@
package config
import (
"errors"
"fmt"
"math"
"reflect"
"slices"
"strconv"
"strings"
lconfig "github.com/lixenwraith/config"
)
// Side is how a plugin meets the network, which decides the tls, auth and acl
// keys it takes
type Side int
const (
Local Side = iota // no network
Listener // the tcp and http sinks
ChainListener // the chain sources, which also bind the node label
Dialer // the chain sinks
)
// Plugin is a catalogue row: a source or sink type and its options table
type Plugin struct {
Role, Type string // "source" or "sink", and the type its config names
Side Side
HTTP bool // speaks HTTP: takes bearer tokens and request limits
Single bool // one instance per process
Summary string
options reflect.Type
}
var plugins = []Plugin{
{"source", "console", Local, false, true, "Standard input, line by line", reflect.TypeFor[ConsoleSourceOptions]()},
{"source", "file", Local, false, false, "Follow the files of a directory that match a pattern", reflect.TypeFor[FileSourceOptions]()},
{"source", "random", Local, false, false, "Random entries, for tests", reflect.TypeFor[RandomSourceOptions]()},
{"source", "null", Local, false, false, "Nothing, ever", reflect.TypeFor[NullSourceOptions]()},
{"source", "tcp_chain", ChainListener, false, false, "Entries from tcp_chain sinks of other instances", reflect.TypeFor[TCPChainSourceOptions]()},
{"source", "http_chain", ChainListener, true, false, "Entry batches from http_chain sinks of other instances", reflect.TypeFor[HTTPChainSourceOptions]()},
{"sink", "console", Local, false, false, "Standard output or standard error", reflect.TypeFor[ConsoleSinkOptions]()},
{"sink", "file", Local, false, false, "Rotating files", reflect.TypeFor[FileSinkOptions]()},
{"sink", "null", Local, false, false, "Discard", reflect.TypeFor[NullSinkOptions]()},
{"sink", "tcp", Listener, false, false, "Serve entries to TCP clients, one per line", reflect.TypeFor[TCPSinkOptions]()},
{"sink", "http", Listener, true, false, "Serve entries as server-sent events, with a browser viewer", reflect.TypeFor[HTTPSinkOptions]()},
{"sink", "tcp_chain", Dialer, false, false, "Forward entries to a tcp_chain source", reflect.TypeFor[TCPChainSinkOptions]()},
{"sink", "http_chain", Dialer, true, false, "Post entry batches to an http_chain source", reflect.TypeFor[HTTPChainSinkOptions]()},
}
// LookupPlugin finds the catalogue row of a source or sink type
func LookupPlugin(role, typ string) (Plugin, bool) {
i := slices.IndexFunc(plugins, func(p Plugin) bool { return p.Role == role && p.Type == typ })
if i < 0 {
return Plugin{}, false
}
return plugins[i], true
}
// KeyError is a value refused at one key, which callers prefix with the path
// of the table holding it
type KeyError struct {
Key string
Err error
}
func (e *KeyError) Error() string { return e.Key + ": " + e.Err.Error() }
func (e *KeyError) Unwrap() error { return e.Err }
// option is one key of an options table, from its field's tags: toml names
// it, default: holds the default as typed on a command line, help: says what
// it does, and lw: lists its rules: required, enum=a|b, min=N, max=N,
// zero=MEANING (zero is a value, not the default), hint=KIND, listener or dialer.
type option struct {
key, def, help string
required bool
enum []string
min, max *float64
zero string
hint, side string
index int
}
// options reads the keys of an options table; a malformed tag panics, as the
// tables are static and every run would fail alike
func options(t reflect.Type) []option {
var opts []option
for i := range t.NumField() {
f := t.Field(i)
key, _, _ := strings.Cut(f.Tag.Get("toml"), ",")
if key == "" || key == "-" {
continue
}
o := option{key: key, def: f.Tag.Get("default"), help: f.Tag.Get("help"), index: i}
for rule := range strings.SplitSeq(f.Tag.Get("lw"), ",") {
name, value, _ := strings.Cut(rule, "=")
switch name {
case "":
case "required":
o.required = true
case "listener", "dialer":
o.side = name
case "enum":
o.enum = strings.Split(value, "|")
case "min", "max":
n, err := strconv.ParseFloat(value, 64)
if err != nil {
panic(fmt.Sprintf("%s.%s: lw %s=%q", t.Name(), key, name, value))
}
if name == "min" {
o.min = &n
} else {
o.max = &n
}
case "zero":
o.zero = value
case "hint":
o.hint = value
default:
panic(fmt.Sprintf("%s.%s: unknown lw rule %q", t.Name(), key, rule))
}
}
opts = append(opts, o)
}
return opts
}
// Decode reads a plugin's config map into its options: unknown keys, kinds,
// the defaults, the tag rules, the tls, auth and acl rules of its side, then the
// options' own Check. It reads no file.
func Decode[T any](role, typ string, m map[string]any) (*T, error) {
v, err := decode(role, typ, m)
if err != nil {
return nil, err
}
return v.(*T), nil
}
func decode(role, typ string, m map[string]any) (any, error) {
p, ok := LookupPlugin(role, typ)
if !ok {
return nil, fmt.Errorf("unknown %s type %q", role, typ)
}
if err := checkKeys(m, p.options, ""); err != nil {
return nil, err
}
m = clone(m)
if err := coerce(p.options, m, ""); err != nil {
return nil, err // a file's wrong kind, named by its key
}
v := reflect.New(p.options)
preset(v.Elem(), m)
if err := lconfig.ScanMap(m, v.Interface()); err != nil {
return nil, err
}
if err := Settle(v.Interface()); err != nil {
return nil, err
}
if n, ok := v.Interface().(networked); ok {
if err := p.checkNetwork(n); err != nil {
return nil, err
}
}
return v.Interface(), nil
}
// Settle gives a decoded table its defaults, applies the tag rules, then its
// own Check: Decode's last steps, for the flow stages a file decodes typed
func Settle(v any) error {
rv := reflect.ValueOf(v).Elem()
settle(rv)
if err := checkTags(rv, ""); err != nil {
return err
}
if c, ok := v.(interface{ Check() error }); ok {
return c.Check()
}
return nil
}
// preset sets every default before decoding, so a switch that defaults on
// stays on unless set off; it does so in the tables m sets
func preset(v reflect.Value, m map[string]any) {
for _, o := range options(v.Type()) {
f := v.Field(o.index)
if sub, ok := m[o.key].(map[string]any); ok && f.Kind() == reflect.Pointer && f.Type().Elem().Kind() == reflect.Struct {
f.Set(reflect.New(f.Type().Elem()))
preset(f.Elem(), sub)
} else if o.def != "" {
setDefault(f, o)
}
}
}
// settle gives each zero key its default, unless zero is a value there or
// the key is a switch, whose off a default would override
func settle(v reflect.Value) {
for _, o := range options(v.Type()) {
f := v.Field(o.index)
switch {
case f.Kind() == reflect.Pointer && f.Type().Elem().Kind() == reflect.Struct:
if !f.IsNil() {
settle(f.Elem())
}
case o.def != "" && o.zero == "" && f.Kind() != reflect.Bool && f.IsZero():
setDefault(f, o)
}
}
}
func setDefault(f reflect.Value, o option) {
var err error
switch f.Kind() {
case reflect.String:
f.SetString(o.def)
case reflect.Int64:
var n int64
n, err = strconv.ParseInt(o.def, 10, 64)
f.SetInt(n)
case reflect.Float64:
var n float64
n, err = strconv.ParseFloat(o.def, 64)
f.SetFloat(n)
case reflect.Bool:
var b bool
b, err = strconv.ParseBool(o.def)
f.SetBool(b)
default:
err = errors.New("no default for this kind")
}
if err != nil {
panic(fmt.Sprintf("%s: default %q: %v", o.key, o.def, err))
}
}
// checkTags applies the tag rules to a table's own keys, naming each by
// prefix and key. A zero number is the default or its zero meaning, and
// meets the range unless required; an empty enum is one only by its zero
// or default.
func checkTags(v reflect.Value, prefix string) error {
for _, o := range options(v.Type()) {
f := v.Field(o.index)
var err error
switch {
case o.required && (f.IsZero() || f.Kind() == reflect.String && strings.TrimSpace(f.String()) == ""):
err = errors.New("must be set")
case f.Kind() == reflect.String && len(o.enum) > 0 && (f.String() != "" || o.zero == "" && o.def == "") && !slices.Contains(o.enum, f.String()):
err = fmt.Errorf("must be one of %s, got %q", strings.Join(o.enum, ", "), f.String())
case f.CanFloat() || f.CanInt():
if f.IsZero() && !o.required {
break
}
n := float64(0)
if f.CanInt() {
n = float64(f.Int())
} else {
n = f.Float()
}
switch {
case math.IsNaN(n) || math.IsInf(n, 0):
err = errors.New("must be a finite number")
case o.min != nil && o.max != nil && !(n >= *o.min && n <= *o.max):
err = fmt.Errorf("must be from %s to %s, got %s", num(*o.min), num(*o.max), num(n))
case o.min != nil && !(n >= *o.min):
err = fmt.Errorf("must be at least %s, got %s", num(*o.min), num(n))
case o.max != nil && !(n <= *o.max):
err = fmt.Errorf("must be at most %s, got %s", num(*o.max), num(n))
}
}
if err != nil {
return &KeyError{prefix + o.key, err}
}
}
return nil
}
func num(n float64) string { return strconv.FormatFloat(n, 'f', -1, 64) }
// sideKeys lists, in order, the keys of a table that apply to one side only
func sideKeys(t reflect.Type, side string) []string {
var keys []string
for _, o := range options(t) {
if o.side == side {
keys = append(keys, o.key)
}
}
return keys
}
// Coerce gives command-line and environment values, which are strings, and
// JSON's floats the kinds a plugin's options declare, so a dump prints them
// typed. Keys and types it does not know are Decode's to refuse.
func Coerce(role, typ string, m map[string]any) error {
p, ok := LookupPlugin(role, typ)
if !ok {
return nil
}
return coerce(p.options, m, "")
}
func coerce(t reflect.Type, m map[string]any, prefix string) error {
for _, o := range options(t) {
value, ok := m[o.key]
if !ok {
continue
}
ft := t.Field(o.index).Type
if ft.Kind() == reflect.Pointer {
ft = ft.Elem()
}
if sub, ok := value.(map[string]any); ok && ft.Kind() == reflect.Struct {
if err := coerce(ft, sub, prefix+o.key+"."); err != nil {
return err
}
continue
}
v, err := coerceValue(ft.Kind(), value)
if err != nil {
return &KeyError{prefix + o.key, err}
}
m[o.key] = v
}
return nil
}
// coerceValue converts text, and a JSON float for an integer; its errors
// never hold the value, which may be a misplaced secret
func coerceValue(kind reflect.Kind, value any) (any, error) {
s, text := value.(string)
switch {
case kind == reflect.Slice && text:
return []any{s}, nil
case kind == reflect.Int64 && text:
if n, err := strconv.ParseInt(s, 10, 64); err == nil {
return n, nil
}
return nil, errors.New("not an integer")
case kind == reflect.Int64:
if f, ok := value.(float64); ok {
if f != math.Trunc(f) || math.Abs(f) > 1<<53 {
return nil, errors.New("not an integer within 2^53")
}
return int64(f), nil
}
case kind == reflect.Float64 && text:
if f, err := strconv.ParseFloat(s, 64); err == nil {
return f, nil
}
return nil, errors.New("not a number")
case kind == reflect.Bool && text:
if b, err := strconv.ParseBool(s); err == nil {
return b, nil
}
return nil, errors.New("not true or false")
}
return value, nil
}
// clone copies m and the tables in it, so decoding leaves the caller's map be
func clone(m map[string]any) map[string]any {
c := make(map[string]any, len(m))
for k, v := range m {
if sub, ok := v.(map[string]any); ok {
v = clone(sub)
}
c[k] = v
}
return c
}
internal/config/schema_test.go
+136
View File
@@ -0,0 +1,136 @@
package config
import (
"math"
"reflect"
"strings"
"testing"
)
// Every catalogue row decodes from its required keys alone: its tags parse,
// its defaults meet its own rules, and an enum's default is one of its values
func TestEveryRowDecodesFromItsRequiredKeys(t *testing.T) {
for _, p := range plugins {
m := map[string]any{}
for _, o := range options(p.options) {
if o.required {
m[o.key] = "1"
}
}
if err := Coerce(p.Role, p.Type, m); err != nil {
t.Fatalf("%s %s: %v", p.Type, p.Role, err)
}
if _, err := decode(p.Role, p.Type, m); err != nil {
t.Errorf("%s %s from %v: %v", p.Type, p.Role, m, err)
}
}
for _, stage := range []any{&HeartbeatConfig{}, &RateLimitConfig{}, &FormatConfig{}, &FilterConfig{}} {
if err := Settle(stage); err != nil {
t.Errorf("%T: %v", stage, err)
}
}
}
// A zero number takes the default unless zero means something there, a switch
// that defaults on stays off when set off, a negative number is refused, and
// what moved from the constructors keeps its meaning: min_disk_free_mb below
// zero is the default, a backoff maximum below the minimum the default maximum
func TestZeroDefaultsAndNegatives(t *testing.T) {
tcp, err := Decode[TCPSinkOptions]("sink", "tcp", map[string]any{"port": int64(1), "buffer_size": int64(0), "keep_alive": false})
if err != nil || tcp.BufferSize != 1000 || tcp.KeepAlive || tcp.Host != "0.0.0.0" {
t.Fatalf("tcp sink: %+v %v", tcp, err)
}
if _, err := Decode[TCPSinkOptions]("sink", "tcp", map[string]any{"port": int64(1), "buffer_size": int64(-1)}); err == nil ||
err.Error() != "buffer_size: must be at least 1, got -1" {
t.Fatalf("negative buffer_size: %v", err)
}
for in, want := range map[any]int64{nil: 0, int64(0): 0, int64(-1): 100, int64(7): 7} {
m := map[string]any{"directory": "d", "name": "n", "min_disk_free_mb": in}
if in == nil {
delete(m, "min_disk_free_mb")
}
file, err := Decode[FileSinkOptions]("sink", "file", m)
if err != nil || file.MinDiskFreeMB != want {
t.Errorf("min_disk_free_mb %v: %+v %v, want %d", in, file, err, want)
}
}
chain, err := Decode[TCPChainSinkOptions]("sink", "tcp_chain", map[string]any{"host": "h", "port": int64(1), "backoff_min_ms": int64(800), "backoff_max_ms": int64(700)})
if err != nil || chain.BackoffMaxMS != 30000 {
t.Fatalf("backoff maximum below the minimum: %+v %v", chain, err)
}
format := FormatConfig{Type: "text"}
if err := Settle(&format); err != nil {
t.Fatalf("format text: %v", err)
}
}
// Command-line text and JSON numbers become the kinds the options declare, so
// a dump prints them typed; a fraction for an integer is refused
func TestCoerceGivesTheDeclaredKinds(t *testing.T) {
m := map[string]any{"port": "8080", "max_connections": float64(3), "keep_alive": "false",
"acl": map[string]any{"allow": "10.0.0.1", "requests_per_second_per_client": "2.5"}, "unknown": "x"}
if err := Coerce("sink", "tcp", m); err != nil {
t.Fatal(err)
}
want := map[string]any{"port": int64(8080), "max_connections": int64(3), "keep_alive": false,
"acl": map[string]any{"allow": []any{"10.0.0.1"}, "requests_per_second_per_client": 2.5}, "unknown": "x"}
if !reflect.DeepEqual(m, want) {
t.Fatalf("coerced %v, want %v", m, want)
}
if err := Coerce("sink", "tcp", map[string]any{"port": 80.5}); err == nil || !strings.HasPrefix(err.Error(), "port: ") {
t.Fatalf("a fraction for an integer: %v", err)
}
}
// ValidateConfig decodes every plugin and flow stage without opening a file,
// and names the path of what it refuses
func TestValidateConfigNamesThePath(t *testing.T) {
valid := func() *Config {
return &Config{Color: "auto", Pipelines: []PipelineConfig{{
Name: "p",
Flow: &FlowConfig{},
PluginSources: []PluginSourceConfig{{ID: "in", Type: "null"}},
PluginSinks: []PluginSinkConfig{{ID: "out", Type: "http", Config: map[string]any{"port": int64(1),
"tls": map[string]any{"enabled": true, "cert_file": "/missing", "key_file": "/missing"}}}},
}}}
}
if err := ValidateConfig(valid()); err != nil {
t.Fatalf("valid configuration, its files missing: %v", err)
}
for want, change := range map[string]func(*Config){
"pipelines[0].plugin_sinks[out].config.port: must be from 1 to 65535, got 70000": func(c *Config) {
c.Pipelines[0].PluginSinks[0].Config["port"] = int64(70000)
},
"pipelines[0].plugin_sinks[out].config: tls: client_auth requires client_ca_file": func(c *Config) {
c.Pipelines[0].PluginSinks[0].Config["tls"].(map[string]any)["client_auth"] = true
},
"pipelines[0].plugin_sinks[out]: duplicate id": func(c *Config) {
c.Pipelines[0].PluginSinks = append(c.Pipelines[0].PluginSinks, PluginSinkConfig{ID: "out", Type: "null"})
},
`pipelines[0].plugin_sources[in].type: unknown source type "nil"`: func(c *Config) {
c.Pipelines[0].PluginSources[0].Type = "nil"
},
"pipelines[0].plugin_sinks[out].config.port: not an integer": func(c *Config) {
c.Pipelines[0].PluginSinks[0].Config["port"] = "http"
},
`pipelines[0].plugin_sinks[up].config: auth: type "mtls" cannot be used with tls.insecure_skip_verify`: func(c *Config) {
c.Pipelines[0].PluginSinks[0] = PluginSinkConfig{ID: "up", Type: "tcp_chain", Config: map[string]any{"host": "h", "port": int64(1),
"tls": map[string]any{"enabled": true, "insecure_skip_verify": true}, "auth": map[string]any{"type": "mtls"}}}
},
"pipelines[0].flow.rate_limit.rate: must be a finite number": func(c *Config) {
c.Pipelines[0].Flow.RateLimit = &RateLimitConfig{Rate: math.Inf(1)}
},
"color: must be one of auto, always, never, got \"\"": func(c *Config) {
c.Color = ""
},
"pipelines[0].flow.heartbeat.interval_ms: must be at least 100, got 10": func(c *Config) {
c.Pipelines[0].Flow.Heartbeat = &HeartbeatConfig{IntervalMS: 10}
},
} {
c := valid()
change(c)
if err := ValidateConfig(c); err == nil || err.Error() != want {
t.Errorf("err = %v, want %s", err, want)
}
}
}
internal/config/spec.go
+2 -2
View File
@@ -44,7 +44,7 @@ var specKinds = []specKind{
}
id, err := instanceID(typ, opts, ids)
p.PluginSources = append(p.PluginSources, PluginSourceConfig{ID: id, Type: typ, Config: opts})
return err
return cmp.Or(err, Coerce("source", typ, opts))
}},
{"sink", true, true, false, func(p *PipelineConfig, typ string, opts map[string]any) error {
ids := make([]string, len(p.PluginSinks))
@@ -53,7 +53,7 @@ var specKinds = []specKind{
}
id, err := instanceID(typ, opts, ids)
p.PluginSinks = append(p.PluginSinks, PluginSinkConfig{ID: id, Type: typ, Config: opts})
return err
return cmp.Or(err, Coerce("sink", typ, opts))
}},
{"filter", true, true, false, func(p *PipelineConfig, typ string, opts map[string]any) error {
opts["type"] = typ
internal/config/spec_test.go
+7 -6
View File
@@ -39,7 +39,7 @@ func TestPipelineSpecGrammar(t *testing.T) {
"source", `file,directory=/var/log/a\,b,pattern=*.log`,
"source", `file,id=app,directory=x\=y`,
"source", `file,directory=c:\\logs\\`,
"sink", "http,port=8080,auth.type=scram,auth.credentials_file=users.toml,tls.cert_file=c",
"sink", "http,port=8080,auth.type=scram,auth.credentials_file=users.toml,tls.enabled=true,tls.cert_file=c,tls.key_file=k",
"filter", "include,patterns=ERROR,patterns=WARN",
"filter", `exclude,patterns=\d{1\,3}`,
"format", `txt,timestamp_format=Jan 2\, 2006`,
@@ -59,9 +59,9 @@ func TestPipelineSpecGrammar(t *testing.T) {
{ID: "file_2", Type: "file", Config: map[string]any{"directory": `c:\logs\`}},
},
PluginSinks: []PluginSinkConfig{{ID: "http", Type: "http", Config: map[string]any{
"port": "8080",
"port": int64(8080),
"auth": map[string]any{"type": "scram", "credentials_file": "users.toml"},
"tls": map[string]any{"cert_file": "c"},
"tls": map[string]any{"enabled": true, "cert_file": "c", "key_file": "k"},
}}},
}, {
Name: "b",
@@ -95,7 +95,8 @@ func TestPipelineSpecGrammar(t *testing.T) {
func TestSpecValueIsOneListEntry(t *testing.T) {
isolateConfig(t)
sink := loadPipelines(t, "source", "null", "sink",
`tcp,auth.type=mtls,auth.allow_patterns=^edge-\d{1\,3}$,auth.allow=CN=a\,O=b,auth.allow=CN=c`)[0].PluginSinks[0]
`tcp,port=9000,tls.enabled=true,tls.self_signed=true,tls.client_auth=true,tls.client_ca_file=ca,`+
`auth.type=mtls,auth.allow_patterns=^edge-\d{1\,3}$,auth.allow=CN=a\,O=b,auth.allow=CN=c`)[0].PluginSinks[0]
var opts TCPSinkOptions
if err := Scan(sink.Config, &opts); err != nil {
t.Fatal(err)
@@ -179,7 +180,7 @@ func TestSpecPipelinesDefaultToStdio(t *testing.T) {
t.Fatalf("pipeline: %+v", p)
}
_, err := Load(Args{Specs: specs("pipeline", "a", "sink", "null", "pipeline", "b", "sink", "null")})
if err == nil || !strings.Contains(err.Error(), `console source already reads stdin in pipeline "a"`) {
if err == nil || !strings.Contains(err.Error(), `one console source may run, and pipeline "a" has it`) {
t.Fatalf("second stdin reader: %v", err)
}
}
@@ -202,7 +203,7 @@ func TestReloadKeepsSpecPipelines(t *testing.T) {
testutil.WriteFile(t, "logwisp.toml", "status_reporter = true\n[[pipelines]]\nname = \"file\"\n")
next, err := m.Reload()
if err != nil || !next.StatusReporter || len(next.Pipelines) != 1 || next.Pipelines[0].Name != "cli" ||
next.Pipelines[0].PluginSources[0].Config["special"] != "true" {
next.Pipelines[0].PluginSources[0].Config["special"] != true {
t.Fatalf("reload: %+v %v", next, err)
}
if err := os.Remove("logwisp.toml"); err != nil {
internal/config/validate.go
+442 -46
View File
@@ -1,21 +1,30 @@
package config
import (
"crypto/sha256"
"encoding/base64"
"errors"
"fmt"
"io/fs"
"maps"
"net/netip"
"os"
"path"
"reflect"
"regexp"
"slices"
"strings"
"github.com/lixenwraith/logwisp/internal/chain"
"github.com/lixenwraith/logwisp/internal/core"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/toml"
)
// ValidateConfig validates top-level structure only
// Value range validation is delegated to component constructors
// ValidateConfig checks a configuration without reading a file it names:
// the settings, then each pipeline's flow stages, and its sources and sinks
// decoded in full, naming the path of a refused key. lw --check opens them.
func ValidateConfig(cfg *Config) error {
if cfg == nil {
return fmt.Errorf("config is nil")
@@ -34,15 +43,21 @@ func ValidateConfig(cfg *Config) error {
names[p.Name] = struct{}{}
}
if err := lconfig.OneOf("auto", "always", "never")(cfg.Color); err != nil {
return fmt.Errorf("color: %w", err)
if err := checkTags(reflect.ValueOf(cfg).Elem(), ""); err != nil {
return err
}
if l := cfg.Logging; l != nil {
if err := checkTags(reflect.ValueOf(l).Elem(), "logging."); err != nil {
return err
}
if l.Console != nil {
if err := checkTags(reflect.ValueOf(l.Console).Elem(), "logging.console."); err != nil {
return err
}
}
}
if err := validateLogConfig(cfg.Logging); err != nil {
return fmt.Errorf("logging: %w", err)
}
stdinReader := ""
single := map[string]string{} // role and type of a single-instance plugin: its pipeline
for i, p := range cfg.Pipelines {
if err := lconfig.NonEmpty(p.Name); err != nil {
return fmt.Errorf("pipeline[%d].name: %w", i, err)
@@ -53,60 +68,441 @@ func ValidateConfig(cfg *Config) error {
if len(p.PluginSinks) == 0 {
return fmt.Errorf("pipeline[%d]: no sinks defined", i)
}
for _, src := range p.PluginSources {
if src.Type != "console" {
continue
if err := checkFlow(p.Flow, fmt.Sprintf("pipelines[%d].flow", i)); err != nil {
return err
}
ids := map[string]bool{}
check := func(role, list, id, typ string, m map[string]any) error {
path := fmt.Sprintf("pipelines[%d].%s[%s]", i, list, id)
plugin, ok := LookupPlugin(role, typ)
switch {
case ids[list+id]:
return fmt.Errorf("%s: duplicate id", path)
case !ok:
return fmt.Errorf("%s.type: unknown %s type %q", path, role, typ)
case plugin.Single && single[role+typ] != "":
return fmt.Errorf("pipeline %q: one %s %s may run, and pipeline %q has it", p.Name, typ, role, single[role+typ])
}
if stdinReader != "" {
return fmt.Errorf("pipeline %q: a console source already reads stdin in pipeline %q", p.Name, stdinReader)
ids[list+id] = true
if plugin.Single {
single[role+typ] = p.Name
}
if _, err := decode(role, typ, m); err != nil {
return at(path+".config", err)
}
return nil
}
for _, s := range p.PluginSources {
if err := check("source", "plugin_sources", s.ID, s.Type, s.Config); err != nil {
return err
}
}
for _, s := range p.PluginSinks {
if err := check("sink", "plugin_sinks", s.ID, s.Type, s.Config); err != nil {
return err
}
stdinReader = p.Name
}
}
return nil
}
// validateLogConfig validates application logging settings
func validateLogConfig(cfg *LogConfig) error {
if cfg == nil {
// checkFlow settles copies of a pipeline's flow stages, which keep their
// defaults for the dump
func checkFlow(f *FlowConfig, path string) error {
if f == nil {
return nil
}
validateOutput := lconfig.OneOf("file", "stdout", "stderr", "split", "all", "none")
if err := validateOutput(cfg.Output); err != nil {
return fmt.Errorf("output: %w", err)
stages := map[string]any{}
if f.Heartbeat != nil {
stages["heartbeat"] = new(*f.Heartbeat)
}
validateLevel := lconfig.OneOf("debug", "info", "warn", "error")
if err := validateLevel(cfg.Level); err != nil {
return fmt.Errorf("level: %w", err)
if f.RateLimit != nil {
stages["rate_limit"] = new(*f.RateLimit)
}
if cfg.Format != "" {
if err := lconfig.OneOf("raw", "txt", "json")(cfg.Format); err != nil {
return fmt.Errorf("format: %w", err)
if f.Format != nil {
stages["format"] = new(*f.Format)
}
for i, filter := range f.Filters {
stages[fmt.Sprintf("filters[%d]", i)] = &filter
}
for _, key := range slices.Sorted(maps.Keys(stages)) {
if err := Settle(stages[key]); err != nil {
return at(path+"."+key, err)
}
}
if cfg.Sanitization != "" {
if err := lconfig.OneOf("raw", "json", "txt", "shell")(cfg.Sanitization); err != nil {
return fmt.Errorf("sanitization: %w", err)
}
}
if cfg.Console != nil {
validateTarget := lconfig.OneOf("stdout", "stderr", "split")
if err := validateTarget(cfg.Console.Target); err != nil {
return fmt.Errorf("console.target: %w", err)
}
}
return nil
}
// Scan decodes a plugin's config map into target, rejecting keys target does
// not declare: a misspelled tls or auth key must fail, not silently disable
// the protection it was meant to configure.
// at prefixes err with the path of the table it was found in
func at(path string, err error) error {
if ke, ok := errors.AsType[*KeyError](err); ok {
return fmt.Errorf("%s.%s: %w", path, ke.Key, ke.Err)
}
return fmt.Errorf("%s: %w", path, err)
}
// networked is the options of a plugin that listens or dials
type networked interface {
network() (host string, tls *TLSOptions, auth *AuthOptions, acl *ACLOptions)
}
func (o *TCPChainSourceOptions) network() (string, *TLSOptions, *AuthOptions, *ACLOptions) {
return o.Host, o.TLS, o.Auth, o.ACL
}
func (o *HTTPChainSourceOptions) network() (string, *TLSOptions, *AuthOptions, *ACLOptions) {
return o.Host, o.TLS, o.Auth, o.ACL
}
func (o *TCPSinkOptions) network() (string, *TLSOptions, *AuthOptions, *ACLOptions) {
return o.Host, o.TLS, o.Auth, o.ACL
}
func (o *HTTPSinkOptions) network() (string, *TLSOptions, *AuthOptions, *ACLOptions) {
return o.Host, o.TLS, o.Auth, o.ACL
}
func (o *TCPChainSinkOptions) network() (string, *TLSOptions, *AuthOptions, *ACLOptions) {
return o.Host, o.TLS, o.Auth, nil
}
func (o *HTTPChainSinkOptions) network() (string, *TLSOptions, *AuthOptions, *ACLOptions) {
return o.Host, o.TLS, o.Auth, nil
}
// checkNetwork applies the host, tls, acl and auth rules of the plugin's side
func (p Plugin) checkNetwork(n networked) error {
host, t, a, acl := n.network()
if _, err := core.Network(host); err != nil {
return &KeyError{"host", err}
}
if err := t.Check(p.Side != Dialer); err != nil {
return err
}
if err := acl.Check(host, p.HTTP, a != nil && len(a.TrustedProxies) > 0); err != nil {
return err
}
return a.Check(p.Side, p.HTTP, t.State())
}
// Check holds the rules between tls keys on a listener or a dialer; a table
// not enabled is ignored whole
func (o *TLSOptions) Check(listener bool) error {
if o == nil || !o.Enabled {
return nil
}
v := reflect.ValueOf(o).Elem()
if err := checkTags(v, "tls."); err != nil {
return err
}
other := "dialer"
if !listener {
other = "listener"
}
for _, opt := range options(v.Type()) {
if f := v.Field(opt.index); opt.side == other && !f.IsZero() && (f.Kind() != reflect.Slice || f.Len() > 0) {
return fmt.Errorf("tls: %s apply to %ss", and(sideKeys(v.Type(), other)), other)
}
}
files, issuer := o.CertFile != "" || o.KeyFile != "", o.IssuerCertFile != "" || o.IssuerKeyFile != ""
switch {
case !listener && o.PinSHA256 != "" && (o.CAFile != "" || o.InsecureSkipVerify):
return fmt.Errorf("tls: pin_sha256 replaces ca_file and insecure_skip_verify: set one")
case !listener:
case files && (issuer || o.SelfSigned) || issuer && o.SelfSigned:
return fmt.Errorf("tls: set one of cert_file and key_file, self_signed, or issuer_cert_file and issuer_key_file")
case len(o.Hosts) > 0 && !issuer && !o.SelfSigned:
return fmt.Errorf("tls: hosts applies to self_signed and issuer certificates")
case issuer && (o.IssuerCertFile == "" || o.IssuerKeyFile == ""):
return fmt.Errorf("tls: issuer_cert_file and issuer_key_file must be set together")
case !files && !issuer && !o.SelfSigned:
return fmt.Errorf("tls: listeners need cert_file and key_file, self_signed, or issuer_cert_file and issuer_key_file")
case o.ClientAuth && o.ClientCAFile == "":
return fmt.Errorf("tls: client_auth requires client_ca_file")
}
if (o.CertFile == "") != (o.KeyFile == "") {
return fmt.Errorf("tls: cert_file and key_file must be set together")
}
if o.PinSHA256 != "" {
_, err := ParsePins(o.PinSHA256)
return err
}
return nil
}
// PinPrefix starts each pin_sha256 entry
const PinPrefix = "sha256//"
// ParsePins reads pin_sha256: SHA-256 hashes of server keys, ';' between several
func ParsePins(s string) ([][]byte, error) {
var pins [][]byte
for p := range strings.SplitSeq(s, ";") {
b64, ok := strings.CutPrefix(strings.TrimSpace(p), PinPrefix)
h, err := base64.StdEncoding.DecodeString(b64)
if !ok || err != nil || len(h) != sha256.Size {
return nil, fmt.Errorf("tls: pin_sha256 %q: want %sBASE64 of a SHA-256, ';' between several", p, PinPrefix)
}
pins = append(pins, h)
}
return pins, nil
}
// TLSState is what the auth rules ask of the tls table beside them
type TLSState struct {
On bool // enabled
ClientAuth bool // a listener verifies client certificates
Unverified bool // a dialer skips verifying the server, and pins nothing instead
}
func (o *TLSOptions) State() TLSState {
if o == nil || !o.Enabled {
return TLSState{}
}
return TLSState{On: true, ClientAuth: o.ClientAuth, Unverified: o.InsecureSkipVerify && o.PinSHA256 == ""}
}
// Check holds the auth rules of a side, http for a plugin that speaks it,
// beside a tls table in state t. It reads no credentials or passwords.
func (o *AuthOptions) Check(side Side, http bool, t TLSState) error {
if o == nil {
return nil
}
if err := checkTags(reflect.ValueOf(o).Elem(), "auth."); err != nil {
return err
}
if o.Type == "" || o.Type == "none" {
// Tuning keys are ignored with the table, but one that names peers or
// credentials means auth was intended and the type was forgotten
if key := o.intentKey(); key != "" {
return fmt.Errorf("auth: %s is set but auth.type is %q", key, "none")
}
return nil
}
switch {
case !t.On && (o.Type != "scram" || len(o.TrustedProxies) == 0):
return fmt.Errorf("auth: type %q requires tls.enabled", o.Type)
case side == Dialer && t.Unverified:
// An unverified server makes identities claims and exposes credentials
return fmt.Errorf("auth: type %q cannot be used with tls.insecure_skip_verify", o.Type)
case side != ChainListener && o.NodeBinding != "" && o.NodeBinding != "none":
return fmt.Errorf("auth: node_binding %q applies only to chain sources", o.NodeBinding)
case o.Type == "mtls" && o.scramKey() != "":
return fmt.Errorf("auth: %s applies only to type %q", o.scramKey(), "scram")
case o.Type == "mtls" && side != Dialer && !t.ClientAuth:
return fmt.Errorf("auth: type %q requires tls.client_auth", "mtls")
case o.Type == "mtls":
for i, pattern := range o.AllowPatterns {
if _, err := regexp.Compile(pattern); err != nil {
return fmt.Errorf("auth: allow_patterns[%d] %q: %w", i, pattern, err)
}
}
return nil
case len(o.Allow) > 0 || len(o.AllowPatterns) > 0:
return fmt.Errorf("auth: allow and allow_patterns apply only to type %q; the credentials file is the allow list", "mtls")
case side == Dialer && (o.CredentialsFile != "" || o.TokenLifetimeMS != 0 || len(o.TrustedProxies) > 0):
return errors.New("auth: credentials_file, token_lifetime_ms and trusted_proxies apply only to listeners")
case side == Dialer && o.Identity != "":
return fmt.Errorf("auth: identity on a dialer pins the server and applies only to type %q", "mtls")
case side == Dialer && (o.Username == "" || o.PasswordFile == ""):
return fmt.Errorf("auth: type %q on a dialer requires username and password_file", "scram")
case side == Dialer:
return nil
case o.Username != "" || o.PasswordFile != "":
return errors.New("auth: username and password_file apply only to dialers")
case o.CredentialsFile == "":
return fmt.Errorf("auth: type %q requires credentials_file", "scram")
case o.TokenLifetimeMS > 0 && !http:
return errors.New("auth: token_lifetime_ms applies only to HTTP listeners")
case len(o.TrustedProxies) > 0 && (side != Listener || !http):
return errors.New("auth: trusted_proxies applies only to the http sink")
case len(o.TrustedProxies) > 0 && o.Identity != "":
return errors.New("auth: identity binds a client certificate, which a TLS-terminating proxy does not pass on; drop it or trusted_proxies")
case o.Identity != "" && !t.ClientAuth:
// Binds the certificate to the user: a peer needs its own of both
return fmt.Errorf("auth: identity under type %q binds the client certificate to the user and requires tls.client_auth", "scram")
}
for _, e := range o.TrustedProxies {
if _, err := ParsePrefix(e, "tcp"); err != nil {
return fmt.Errorf("auth: trusted_proxies entry %q: %w", e, err)
}
}
return nil
}
// intentKey names the first key that only makes sense with authentication on
func (o *AuthOptions) intentKey() string {
switch {
case len(o.Allow) > 0:
return "allow"
case len(o.AllowPatterns) > 0:
return "allow_patterns"
}
return o.scramKey()
}
func (o *AuthOptions) scramKey() string {
switch {
case o.CredentialsFile != "":
return "credentials_file"
case o.TokenLifetimeMS != 0:
return "token_lifetime_ms"
case o.Username != "":
return "username"
case o.PasswordFile != "":
return "password_file"
case len(o.TrustedProxies) > 0:
return "trusted_proxies"
}
return ""
}
// Check holds the acl rules of a listener on host, http for one serving
// requests, proxied for one behind auth.trusted_proxies, whose rules take
// clients of either family, as does a PROXY header
func (o *ACLOptions) Check(host string, http, proxied bool) error {
if o == nil {
return nil
}
if err := checkTags(reflect.ValueOf(o).Elem(), "acl."); err != nil {
return err
}
header := o.ProxyProtocol == "optional" || o.ProxyProtocol == "required"
rps := o.RequestsPerSecondPerClient
switch {
case header && len(o.ProxyFrom) == 0:
return fmt.Errorf("acl: proxy_protocol %s needs proxy_from, the proxies that may send the header", o.ProxyProtocol)
case !header && len(o.ProxyFrom) > 0:
return errors.New("acl: proxy_from needs proxy_protocol optional or required")
case rps > 0 && !http:
return errors.New("acl: requests_per_second_per_client applies only to HTTP listeners")
case o.MaxConnectionsPerClient > 0 && proxied:
return errors.New("acl: max_connections_per_client counts connections, behind trusted_proxies the proxy's; cap clients at the proxy")
}
network, err := core.Network(host)
if err != nil {
return err
}
clients := network
if header || proxied {
clients = "tcp"
}
for _, list := range []struct {
key, network string
entries []string
}{{"proxy_from", network, o.ProxyFrom}, {"allow", clients, o.Allow}, {"deny", clients, o.Deny}} {
for _, e := range list.entries {
if _, err := ParsePrefix(e, list.network); err != nil {
return fmt.Errorf("acl: %s entry %q: %w", list.key, e, err)
}
}
}
return nil
}
// ParsePrefix reads an address or CIDR of network's family ("tcp": either)
func ParsePrefix(entry, network string) (netip.Prefix, error) {
entry = strings.TrimSpace(entry)
prefix, err := netip.ParsePrefix(entry)
if err != nil {
addr, aerr := netip.ParseAddr(entry)
if aerr != nil || addr.Zone() != "" {
return netip.Prefix{}, fmt.Errorf("neither an address nor a CIDR (and no zone)")
}
prefix = netip.PrefixFrom(addr, addr.BitLen())
}
switch a := prefix.Addr(); {
case a.Is4In6():
return netip.Prefix{}, fmt.Errorf("IPv4-mapped; write it as IPv4")
case network == "tcp4" && !a.Is4(), network == "tcp6" && a.Is4():
return netip.Prefix{}, fmt.Errorf("not of the listener's family (%s)", network)
}
return prefix, nil
}
// and joins keys as a sentence does: "a, b and c"
func and(keys []string) string {
if len(keys) < 2 {
return strings.Join(keys, "")
}
return strings.Join(keys[:len(keys)-1], ", ") + " and " + keys[len(keys)-1]
}
// Check keeps a stream and a status path ServeMux and a URL take literally,
// apart, and off the authentication paths; a login page needs browsers that
// log in, behind a proxy
func (o *HTTPSinkOptions) Check() error {
for _, p := range []struct{ key, path string }{{"stream_path", o.StreamPath}, {"status_path", o.StatusPath}} {
c := path.Clean(p.path)
switch {
case !strings.HasPrefix(p.path, "/") || strings.ContainsAny(p.path, "{}%?#") || c != p.path && (c == "/" || c+"/" != p.path):
return &KeyError{p.key, fmt.Errorf("%q must start with '/', hold no '//', '.' or '..' segment, and none of '{', '}', '%%', '?', '#'", p.path)}
case p.path == chain.AuthPath || strings.HasPrefix(p.path, chain.AuthPath+"/"):
return &KeyError{p.key, fmt.Errorf("%s and the paths under it are reserved for authentication", chain.AuthPath)}
}
}
behind := o.Auth != nil && o.Auth.Type == "scram" && len(o.Auth.TrustedProxies) > 0
switch {
case o.StreamPath == o.StatusPath:
return fmt.Errorf("stream_path and status_path must differ")
case (o.LoginPage || o.ViewerPage) && !behind:
return errors.New("login_page and viewer_page apply to scram behind auth.trusted_proxies, where browsers log in; without auth or under mtls the viewer is always served")
case o.ViewerPage && !o.LoginPage:
return errors.New("viewer_page needs login_page, where it sends a signed-out viewer")
}
return nil
}
func (o *HTTPChainSourceOptions) Check() error { return checkIngestPath(o.IngestPath) }
func (o *HTTPChainSinkOptions) Check() error {
o.BackoffMaxMS = backoffMax(o.BackoffMinMS, o.BackoffMaxMS)
return checkIngestPath(o.IngestPath)
}
func checkIngestPath(p string) error {
switch {
case !strings.HasPrefix(p, "/"):
return &KeyError{"ingest_path", errors.New("must start with '/'")}
case p == chain.AuthPath:
return &KeyError{"ingest_path", fmt.Errorf("%s is reserved for authentication", chain.AuthPath)}
}
return nil
}
func (o *TCPChainSinkOptions) Check() error {
o.BackoffMaxMS = backoffMax(o.BackoffMinMS, o.BackoffMaxMS)
return nil
}
// backoffMax replaces a maximum below the minimum with the default maximum
func backoffMax(lo, hi int64) int64 {
if hi < lo {
return 30000
}
return hi
}
// Check reads a negative min_disk_free_mb as 100; 0, or none set, keeps no space free
func (o *FileSinkOptions) Check() error {
if o.MinDiskFreeMB < 0 {
o.MinDiskFreeMB = 100
}
return nil
}
// Check caps the jitter at the interval
func (o *RandomSourceOptions) Check() error {
o.JitterMS = min(o.JitterMS, o.IntervalMS)
return nil
}
// Check compiles the patterns
func (c *FilterConfig) Check() error {
for i, pattern := range c.Patterns {
if _, err := regexp.Compile(pattern); err != nil {
return &KeyError{fmt.Sprintf("patterns[%d]", i), fmt.Errorf("%q: %w", pattern, err)}
}
}
return nil
}
// Scan decodes a flow stage's command-line map into target, rejecting keys
// target does not declare: a misspelled key must fail, not be dropped.
func Scan(configMap map[string]any, target any) error {
if err := checkKeys(configMap, reflect.TypeOf(target), ""); err != nil {
return err
internal/filter/filter.go
+2 -21
View File
@@ -9,7 +9,6 @@ import (
"github.com/lixenwraith/logwisp/internal/config"
"github.com/lixenwraith/logwisp/internal/core"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -28,26 +27,8 @@ type Filter struct {
// NewFilter creates a new filter from a configuration
func NewFilter(cfg config.FilterConfig, logger *log.Logger) (*Filter, error) {
// Validate enums before setting defaults
if cfg.Type != "" {
validateType := lconfig.OneOf(config.FilterTypeInclude, config.FilterTypeExclude)
if err := validateType(cfg.Type); err != nil {
return nil, fmt.Errorf("type: %w", err)
}
}
if cfg.Logic != "" {
validateLogic := lconfig.OneOf(config.FilterLogicOr, config.FilterLogicAnd)
if err := validateLogic(cfg.Logic); err != nil {
return nil, fmt.Errorf("logic: %w", err)
}
}
// Set defaults
if cfg.Type == "" {
cfg.Type = config.FilterTypeInclude
}
if cfg.Logic == "" {
cfg.Logic = config.FilterLogicOr
if err := config.Settle(&cfg); err != nil {
return nil, err
}
f := &Filter{
internal/flow/heartbeat.go
+2 -23
View File
@@ -3,7 +3,6 @@ package flow
import (
"context"
"encoding/json"
"fmt"
"strconv"
"sync/atomic"
"time"
@@ -12,17 +11,10 @@ import (
"github.com/lixenwraith/logwisp/internal/core"
"github.com/lixenwraith/logwisp/internal/format"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
"github.com/lixenwraith/log/formatter"
)
const (
MinHeartbeatIntervalMS = 100
DefaultHeartbeatIntervalMS = 1000
DefaultHeartbeatFormat = "txt"
)
// HeartbeatGenerator produces periodic heartbeat events
type HeartbeatGenerator struct {
config *config.HeartbeatConfig
@@ -38,21 +30,8 @@ func NewHeartbeatGenerator(cfg *config.HeartbeatConfig, formatter format.Formatt
return nil, nil
}
// Validate
if cfg.IntervalMS == 0 {
cfg.IntervalMS = DefaultHeartbeatIntervalMS
} else if cfg.IntervalMS < MinHeartbeatIntervalMS {
return nil, fmt.Errorf("interval_ms: must be >= %d, got %d", MinHeartbeatIntervalMS, cfg.IntervalMS)
}
validateFormat := lconfig.OneOf("txt", "json", "raw", "")
if err := validateFormat(cfg.Format); err != nil {
return nil, fmt.Errorf("format: %w", err)
}
// Defaults
if cfg.Format == "" {
cfg.Format = DefaultHeartbeatFormat
if err := config.Settle(cfg); err != nil {
return nil, err
}
hg := &HeartbeatGenerator{
internal/flow/ratelimiter.go
+7 -27
View File
@@ -1,15 +1,12 @@
package flow
import (
"fmt"
"strings"
"sync/atomic"
"github.com/lixenwraith/logwisp/internal/config"
"github.com/lixenwraith/logwisp/internal/core"
"github.com/lixenwraith/logwisp/internal/tokenbucket"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -27,36 +24,19 @@ type RateLimiter struct {
// NewRateLimiter creates a new pipeline-level rate limiter from configuration
func NewRateLimiter(cfg config.RateLimitConfig, logger *log.Logger) (*RateLimiter, error) {
// Rate <= 0 means disabled
if cfg.Rate <= 0 {
return nil, nil // No rate limit
if err := config.Settle(&cfg); err != nil {
return nil, err
}
// Validate
if err := lconfig.NonNegative(cfg.Rate); err != nil {
return nil, fmt.Errorf("rate: %w", err)
if cfg.Rate == 0 {
return nil, nil // off
}
if err := lconfig.NonNegative(cfg.Burst); err != nil {
return nil, fmt.Errorf("burst: %w", err)
}
if err := lconfig.NonNegative(cfg.MaxEntrySizeBytes); err != nil {
return nil, fmt.Errorf("max_entry_size_bytes: %w", err)
}
// Defaults
burst := cfg.Burst
if burst <= 0 {
if burst == 0 {
burst = cfg.Rate
}
var policy config.RateLimitPolicy
switch strings.ToLower(cfg.Policy) {
case "drop":
policy := config.PolicyPass
if cfg.Policy == "drop" {
policy = config.PolicyDrop
case "pass", "":
policy = config.PolicyPass
default:
return nil, fmt.Errorf("policy: must be one of [drop, pass], got %s", cfg.Policy)
}
l := &RateLimiter{
internal/format/adapter.go
+2 -24
View File
@@ -3,21 +3,15 @@ package format
import (
"bytes"
"encoding/json"
"fmt"
"sync"
"github.com/lixenwraith/logwisp/internal/config"
"github.com/lixenwraith/logwisp/internal/core"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log/formatter"
"github.com/lixenwraith/log/sanitizer"
)
const (
DefaultFormatType = "raw"
)
// FormatterAdapter wraps log/formatter for logwisp compatibility
type FormatterAdapter struct {
formatter *formatter.Formatter
@@ -28,24 +22,8 @@ type FormatterAdapter struct {
// NewFormatterAdapter creates adapter from config
func NewFormatterAdapter(cfg *config.FormatConfig) (*FormatterAdapter, error) {
// Validate
if cfg.Type != "" {
validateType := lconfig.OneOf("json", "txt", "text", "raw")
if err := validateType(cfg.Type); err != nil {
return nil, fmt.Errorf("type: %w", err)
}
}
if cfg.SanitizerPolicy != "" {
validatePolicy := lconfig.OneOf("raw", "json", "txt", "shell")
if err := validatePolicy(cfg.SanitizerPolicy); err != nil {
return nil, fmt.Errorf("sanitizer_policy: %w", err)
}
}
// Defaults
if cfg.Type == "" {
cfg.Type = DefaultFormatType
if err := config.Settle(cfg); err != nil {
return nil, err
}
// Create sanitizer based on policy
internal/format/format.go
+1 -5
View File
@@ -17,11 +17,7 @@ type Formatter interface {
// NewFormatter creates a Formatter using formatter/sanitizer packages
func NewFormatter(cfg *config.FormatConfig) (Formatter, error) {
if cfg == nil {
cfg = &config.FormatConfig{
Type: DefaultFormatType,
Flags: 0,
SanitizerPolicy: "raw",
}
cfg = &config.FormatConfig{}
}
return NewFormatterAdapter(cfg)
internal/netacl/netacl.go
+5 -37
View File
@@ -88,24 +88,12 @@ func New(o *config.ACLOptions, host string, kind Kind, l *log.Logger, component,
if o == nil {
return nil, nil
}
if err := o.Check(host, kind != TCP, kind == HTTPProxied); err != nil {
return nil, err
}
proxied := o.ProxyProtocol == "optional" || o.ProxyProtocol == "required"
rps := o.RequestsPerSecondPerClient
switch {
case !proxied && o.ProxyProtocol != "" && o.ProxyProtocol != "off":
return nil, fmt.Errorf("acl: proxy_protocol %q is none of off, optional and required", o.ProxyProtocol)
case proxied && len(o.ProxyFrom) == 0:
return nil, fmt.Errorf("acl: proxy_protocol %s needs proxy_from, the proxies that may send the header", o.ProxyProtocol)
case !proxied && len(o.ProxyFrom) > 0:
return nil, errors.New("acl: proxy_from needs proxy_protocol optional or required")
case o.MaxConnectionsPerClient < 0:
return nil, fmt.Errorf("acl: max_connections_per_client %d is negative", o.MaxConnectionsPerClient)
case !(rps >= 0) || math.IsInf(rps, 0):
return nil, fmt.Errorf("acl: requests_per_second_per_client %v is not a rate", rps)
case rps > 0 && kind == TCP:
return nil, errors.New("acl: requests_per_second_per_client applies only to HTTP listeners")
case o.MaxConnectionsPerClient > 0 && kind == HTTPProxied:
return nil, errors.New("acl: max_connections_per_client counts connections, behind trusted_proxies the proxy's; cap clients at the proxy")
case len(o.Allow)+len(o.Deny)+len(o.ProxyFrom) == 0 && o.MaxConnectionsPerClient == 0 && rps == 0:
if len(o.Allow)+len(o.Deny)+len(o.ProxyFrom) == 0 && o.MaxConnectionsPerClient == 0 && rps == 0 {
return nil, nil
}
network, err := core.Network(host)
@@ -116,7 +104,7 @@ func New(o *config.ACLOptions, host string, kind Kind, l *log.Logger, component,
compile := func(key, network string, entries []string) ([]netip.Prefix, error) {
var rules []netip.Prefix
for _, e := range entries {
prefix, err := Parse(e, network)
prefix, err := config.ParsePrefix(e, network)
if err != nil {
return nil, fmt.Errorf("acl: %s entry %q: %w", key, e, err)
}
@@ -150,26 +138,6 @@ func New(o *config.ACLOptions, host string, kind Kind, l *log.Logger, component,
return p, nil
}
// Parse reads an address or CIDR of network's family ("tcp": either)
func Parse(entry, network string) (netip.Prefix, error) {
entry = strings.TrimSpace(entry)
prefix, err := netip.ParsePrefix(entry)
if err != nil {
addr, aerr := netip.ParseAddr(entry)
if aerr != nil || addr.Zone() != "" {
return netip.Prefix{}, fmt.Errorf("neither an address nor a CIDR (and no zone)")
}
prefix = netip.PrefixFrom(addr, addr.BitLen())
}
switch a := prefix.Addr(); {
case a.Is4In6():
return netip.Prefix{}, fmt.Errorf("IPv4-mapped; write it as IPv4")
case network == "tcp4" && !a.Is4(), network == "tcp6" && a.Is4():
return netip.Prefix{}, fmt.Errorf("not of the listener's family (%s)", network)
}
return prefix, nil
}
// ParseHost reads an address with or without a port, an IPv4-mapped one as IPv4
func ParseHost(s string) (netip.Addr, bool) {
if ap, err := netip.ParseAddrPort(s); err == nil {
internal/pipeline/pipeline_test.go
+6 -4
View File
@@ -69,12 +69,14 @@ func (s *slowSink) GetStats() sink.SinkStats { return sink.SinkStats{} }
var testSinks = map[string]*slowSink{}
// The fakes take catalogue types that no plugin package registers in this
// test binary
func init() {
plugin.RegisterSource("test_finite", func(id string, cfg map[string]any, _ *log.Logger, _ *session.Proxy) (source.Source, error) {
plugin.RegisterSource("random", func(id string, cfg map[string]any, _ *log.Logger, _ *session.Proxy) (source.Source, error) {
n, _ := strconv.Atoi(fmt.Sprint(cfg["n"]))
return &finiteSource{n: n}, nil
})
plugin.RegisterSink("test_slow", func(id string, _ map[string]any, _ *log.Logger, _ *session.Proxy) (sink.Sink, error) {
plugin.RegisterSink("null", func(id string, _ map[string]any, _ *log.Logger, _ *session.Proxy) (sink.Sink, error) {
return testSinks[id], nil
})
}
@@ -84,8 +86,8 @@ func newTestPipeline(t *testing.T, n int, s *slowSink) *Pipeline {
testSinks[t.Name()] = s
p, err := NewPipeline(&config.PipelineConfig{
Name: t.Name(),
PluginSources: []config.PluginSourceConfig{{ID: "src", Type: "test_finite", Config: map[string]any{"n": n}}},
PluginSinks: []config.PluginSinkConfig{{ID: t.Name(), Type: "test_slow"}},
PluginSources: []config.PluginSourceConfig{{ID: "src", Type: "random", Config: map[string]any{"n": n}}},
PluginSinks: []config.PluginSinkConfig{{ID: t.Name(), Type: "null"}},
}, log.NewLogger())
if err != nil {
t.Fatal(err)
internal/pipeline/registry.go
+9 -8
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"sync"
"github.com/lixenwraith/logwisp/internal/config"
"github.com/lixenwraith/logwisp/internal/plugin"
"github.com/lixenwraith/logwisp/internal/session"
"github.com/lixenwraith/logwisp/internal/sink"
@@ -59,7 +60,7 @@ func NewRegistry(pipelineName string, logger *log.Logger) *Registry {
func (r *Registry) CreateSource(
id string,
pluginType string,
config map[string]any,
options map[string]any,
logger *log.Logger,
proxy *session.Proxy,
) (source.Source, error) {
@@ -72,8 +73,8 @@ func (r *Registry) CreateSource(
}
// Check single instance constraint
if meta, ok := plugin.GetSourceMetadata(pluginType); ok {
if meta.MaxInstances == 1 && r.sourceTypeCounts[pluginType] >= 1 {
if row, ok := config.LookupPlugin("source", pluginType); ok {
if row.Single && r.sourceTypeCounts[pluginType] >= 1 {
return nil, fmt.Errorf("source type %s only allows single instance", pluginType)
}
}
@@ -85,7 +86,7 @@ func (r *Registry) CreateSource(
}
// Create instance
src, err := constructor(id, config, logger, proxy)
src, err := constructor(id, options, logger, proxy)
if err != nil {
return nil, fmt.Errorf("failed to create source %s: %w", id, err)
}
@@ -106,7 +107,7 @@ func (r *Registry) CreateSource(
func (r *Registry) CreateSink(
id string,
pluginType string,
config map[string]any,
options map[string]any,
logger *log.Logger,
proxy *session.Proxy,
) (sink.Sink, error) {
@@ -119,8 +120,8 @@ func (r *Registry) CreateSink(
}
// Check single instance constraint
if meta, ok := plugin.GetSinkMetadata(pluginType); ok {
if meta.MaxInstances == 1 && r.sinkTypeCounts[pluginType] >= 1 {
if row, ok := config.LookupPlugin("sink", pluginType); ok {
if row.Single && r.sinkTypeCounts[pluginType] >= 1 {
return nil, fmt.Errorf("sink type %s only allows single instance", pluginType)
}
}
@@ -132,7 +133,7 @@ func (r *Registry) CreateSink(
}
// Create instance
snk, err := constructor(id, config, logger, proxy)
snk, err := constructor(id, options, logger, proxy)
if err != nil {
return nil, fmt.Errorf("failed to create sink %s: %w", id, err)
}
internal/plugin/factory.go
+11 -70
View File
@@ -4,7 +4,7 @@ import (
"fmt"
"sync"
"github.com/lixenwraith/logwisp/internal/core"
"github.com/lixenwraith/logwisp/internal/config"
"github.com/lixenwraith/logwisp/internal/session"
"github.com/lixenwraith/logwisp/internal/sink"
"github.com/lixenwraith/logwisp/internal/source"
@@ -28,18 +28,10 @@ type SinkFactory func(
sessions *session.Proxy,
) (sink.Sink, error)
// PluginMetadata stores metadata about a plugin type
type PluginMetadata struct {
Capabilities []core.Capability
MaxInstances int // 0 = unlimited, 1 = single instance only
}
// registry encapsulates all plugin factories with lazy initialization
type registry struct {
sourceFactories map[string]SourceFactory
sinkFactories map[string]SinkFactory
sourceMetadata map[string]*PluginMetadata
sinkMetadata map[string]*PluginMetadata
mu sync.RWMutex
}
@@ -54,15 +46,17 @@ func getRegistry() *registry {
globalRegistry = &registry{
sourceFactories: make(map[string]SourceFactory),
sinkFactories: make(map[string]SinkFactory),
sourceMetadata: make(map[string]*PluginMetadata),
sinkMetadata: make(map[string]*PluginMetadata),
}
})
return globalRegistry
}
// RegisterSource registers a source factory function
// RegisterSource registers a source factory function for a type the
// catalogue (config.LookupPlugin) has a row for
func RegisterSource(name string, constructor SourceFactory) error {
if _, ok := config.LookupPlugin("source", name); !ok {
return fmt.Errorf("source type %s has no catalogue row", name)
}
r := getRegistry()
r.mu.Lock()
defer r.mu.Unlock()
@@ -71,17 +65,15 @@ func RegisterSource(name string, constructor SourceFactory) error {
return fmt.Errorf("source type %s already registered", name)
}
r.sourceFactories[name] = constructor
// Set default metadata
r.sourceMetadata[name] = &PluginMetadata{
MaxInstances: 0, // Unlimited by default
}
return nil
}
// RegisterSink registers a sink factory function
// RegisterSink registers a sink factory function for a type the catalogue
// has a row for
func RegisterSink(name string, constructor SinkFactory) error {
if _, ok := config.LookupPlugin("sink", name); !ok {
return fmt.Errorf("sink type %s has no catalogue row", name)
}
r := getRegistry()
r.mu.Lock()
defer r.mu.Unlock()
@@ -90,39 +82,6 @@ func RegisterSink(name string, constructor SinkFactory) error {
return fmt.Errorf("sink type %s already registered", name)
}
r.sinkFactories[name] = constructor
// Set default metadata
r.sinkMetadata[name] = &PluginMetadata{
MaxInstances: 0, // Unlimited by default
}
return nil
}
// SetSourceMetadata sets metadata for a source type (call after RegisterSource)
func SetSourceMetadata(name string, metadata *PluginMetadata) error {
r := getRegistry()
r.mu.Lock()
defer r.mu.Unlock()
if _, exists := r.sourceFactories[name]; !exists {
return fmt.Errorf("source type %s not registered", name)
}
r.sourceMetadata[name] = metadata
return nil
}
// SetSinkMetadata sets metadata for a sink type (call after RegisterSink)
func SetSinkMetadata(name string, metadata *PluginMetadata) error {
r := getRegistry()
r.mu.Lock()
defer r.mu.Unlock()
if _, exists := r.sinkFactories[name]; !exists {
return fmt.Errorf("sink type %s not registered", name)
}
r.sinkMetadata[name] = metadata
return nil
}
@@ -144,24 +103,6 @@ func GetSink(name string) (SinkFactory, bool) {
return constructor, exists
}
// GetSourceMetadata retrieves metadata for a source type
func GetSourceMetadata(name string) (*PluginMetadata, bool) {
r := getRegistry()
r.mu.RLock()
defer r.mu.RUnlock()
meta, exists := r.sourceMetadata[name]
return meta, exists
}
// GetSinkMetadata retrieves metadata for a sink type
func GetSinkMetadata(name string) (*PluginMetadata, bool) {
r := getRegistry()
r.mu.RLock()
defer r.mu.RUnlock()
meta, exists := r.sinkMetadata[name]
return meta, exists
}
// ListSources returns all registered source types
func ListSources() []string {
r := getRegistry()
internal/sink/console/console.go
+3 -28
View File
@@ -19,7 +19,6 @@ import (
"github.com/lixenwraith/logwisp/internal/sink"
"github.com/lixenwraith/color"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
"github.com/lixenwraith/terminal/inline"
"golang.org/x/term"
@@ -61,14 +60,6 @@ type ConsoleSink struct {
lastProcessed atomic.Value // time.Time
}
const (
// Defaults
DefaultConsoleTarget = "stdout"
DefaultConsoleBufferSize = 1000
DefaultConsoleEscape = "auto"
DefaultConsoleColor = "auto"
)
// levelStyles colors the levels in ANSI 16, so the terminal's theme picks the
// shades and a text console renders them; cyan, not blue, stays readable on black
var levelStyles = map[string]inline.Style{
@@ -86,21 +77,9 @@ func NewConsoleSinkPlugin(
logger *log.Logger,
proxy *session.Proxy,
) (sink.Sink, error) {
opts := &config.ConsoleSinkOptions{Target: DefaultConsoleTarget, Escape: DefaultConsoleEscape, Color: DefaultConsoleColor}
// Scan config map into struct
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
}
if err := lconfig.OneOf("stdout", "stderr")(opts.Target); err != nil {
return nil, fmt.Errorf("target: %w", err)
}
if err := lconfig.OneOf("auto", "always", "never")(opts.Escape); err != nil {
return nil, fmt.Errorf("escape: %w", err)
}
if err := lconfig.OneOf("auto", "always", "never")(opts.Color); err != nil {
return nil, fmt.Errorf("color: %w", err)
opts, err := config.Decode[config.ConsoleSinkOptions]("sink", "console", configMap)
if err != nil {
return nil, err
}
output := os.Stdout
@@ -108,10 +87,6 @@ func NewConsoleSinkPlugin(
output = os.Stderr
}
if opts.BufferSize <= 0 {
opts.BufferSize = DefaultConsoleBufferSize
}
// inline decides auto: a terminal, NO_COLOR unset, TERM not dumb. Paint
// returns its text unchanged exactly when color is off.
printer := inline.New(output)
internal/sink/file/file.go
+3 -45
View File
@@ -12,7 +12,6 @@ import (
"github.com/lixenwraith/logwisp/internal/session"
"github.com/lixenwraith/logwisp/internal/sink"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -50,16 +49,6 @@ type FileSink struct {
lastProcessed atomic.Value // time.Time
}
const (
// Defaults
DefaultFileMaxSizeMB = 100
DefaultFileMaxTotalSizeMB = 1000
DefaultFileMinDiskFreeMB = 100
DefaultFileRetentionHours = 168 // 7 days
DefaultFileBufferSize = 1000
DefaultFileFlushIntervalMs = 100
)
// NewFileSinkPlugin creates a file sink through plugin factory
func NewFileSinkPlugin(
id string,
@@ -68,42 +57,11 @@ func NewFileSinkPlugin(
proxy *session.Proxy,
) (sink.Sink, error) {
// Create empty config struct
opts := &config.FileSinkOptions{}
// Scan config map into struct
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
opts, err := config.Decode[config.FileSinkOptions]("sink", "file", configMap)
if err != nil {
return nil, err
}
// Validate
if err := lconfig.NonEmpty(opts.Directory); err != nil {
return nil, fmt.Errorf("directory: %w", err)
}
if err := lconfig.NonEmpty(opts.Name); err != nil {
return nil, fmt.Errorf("name: %w", err)
}
// Defaults
if opts.MaxSizeMB <= 0 {
opts.MaxSizeMB = DefaultFileMaxSizeMB
}
if opts.MaxTotalSizeMB <= 0 {
opts.MaxTotalSizeMB = DefaultFileMaxTotalSizeMB
}
if opts.MinDiskFreeMB < 0 {
opts.MinDiskFreeMB = DefaultFileMinDiskFreeMB
}
if opts.RetentionHours <= 0 {
opts.RetentionHours = DefaultFileRetentionHours
}
if opts.BufferSize <= 0 {
opts.BufferSize = DefaultFileBufferSize
}
if opts.FlushIntervalMs <= 0 {
opts.FlushIntervalMs = DefaultFileFlushIntervalMs
}
// Create configuration for the internal log writer
writerConfig := log.DefaultConfig()
writerConfig.Directory = opts.Directory
writerConfig.Name = opts.Name
internal/sink/http/http.go
+6 -57
View File
@@ -10,7 +10,6 @@ import (
"maps"
"net"
"net/http"
"path"
"slices"
"strconv"
"strings"
@@ -29,7 +28,6 @@ import (
"github.com/lixenwraith/logwisp/internal/tlsx"
"github.com/lixenwraith/logwisp/internal/version"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -39,14 +37,7 @@ func init() {
}
}
const (
DefaultHTTPHost = "0.0.0.0"
DefaultHTTPBufferSize = 1000
DefaultHTTPClientBufferSize = 256
DefaultHTTPStreamPath = "/stream"
DefaultHTTPStatusPath = "/status"
HTTPReadHeaderTimeout = 10 * time.Second
)
const HTTPReadHeaderTimeout = 10 * time.Second
// HTTPSink streams log entries via Server-Sent Events
// Server.WriteTimeout is deliberately unset (it would terminate long-lived SSE streams)
@@ -117,42 +108,13 @@ func NewHTTPSinkPlugin(
logger *log.Logger,
proxy *session.Proxy,
) (sink.Sink, error) {
opts := &config.HTTPSinkOptions{
Host: DefaultHTTPHost,
WriteTimeoutMS: 0, // SSE indefinite streaming
}
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
}
if err := lconfig.Port(opts.Port); err != nil {
return nil, fmt.Errorf("port: %w", err)
opts, err := config.Decode[config.HTTPSinkOptions]("sink", "http", configMap)
if err != nil {
return nil, err
}
network, err := core.Network(opts.Host)
if err != nil {
return nil, fmt.Errorf("host: %w", err)
}
if opts.StreamPath == "" {
opts.StreamPath = DefaultHTTPStreamPath
}
if opts.StatusPath == "" {
opts.StatusPath = DefaultHTTPStatusPath
}
for _, o := range []struct{ name, p string }{{"stream_path", opts.StreamPath}, {"status_path", opts.StatusPath}} {
if !routable(o.p) {
return nil, fmt.Errorf("%s %q: must start with '/', hold no '//', '.' or '..' segment, and none of '{', '}', '%%', '?', '#'", o.name, o.p)
}
if p := o.p; p == chain.AuthPath || strings.HasPrefix(p, chain.AuthPath+"/") {
return nil, fmt.Errorf("%s and the paths under it are reserved for authentication", chain.AuthPath)
}
}
if opts.StreamPath == opts.StatusPath {
return nil, fmt.Errorf("stream_path and status_path must differ")
}
if opts.BufferSize <= 0 {
opts.BufferSize = DefaultHTTPBufferSize
}
if opts.ClientBufferSize <= 0 {
opts.ClientBufferSize = DefaultHTTPClientBufferSize
return nil, err
}
tlsCfg, err := tlsx.Server(opts.TLS, opts.Host)
if err != nil {
@@ -166,16 +128,10 @@ func NewHTTPSinkPlugin(
if err != nil {
return nil, err
}
authPolicy, err := authz.New(opts.Auth, tlsCfg, aclPolicy, authz.RoleListener, authz.HTTP)
authPolicy, err := authz.New(opts.Auth, tlsCfg, aclPolicy, config.Listener, authz.HTTP)
if err != nil {
return nil, err
}
switch {
case (opts.LoginPage || opts.ViewerPage) && !authPolicy.BehindProxy():
return nil, errors.New("login_page and viewer_page apply to scram behind auth.trusted_proxies, where browsers log in; without auth or under mtls the viewer is always served")
case opts.ViewerPage && !opts.LoginPage:
return nil, errors.New("viewer_page needs login_page, where it sends a signed-out viewer")
}
web, err := webHandlers(opts, authPolicy)
if err != nil {
return nil, err
@@ -705,13 +661,6 @@ func (h *HTTPSink) handleAuth(w http.ResponseWriter, r *http.Request) {
}
}
// routable reports a path ServeMux and a URL take literally: clean, a trailing
// '/' aside, without the braces of wildcards, an escape, a query or a fragment
func routable(p string) bool {
c := path.Clean(p)
return strings.HasPrefix(p, "/") && !strings.ContainsAny(p, "{}%?#") && (c == p || c != "/" && c+"/" == p)
}
// exact keeps a path ending in '/' from matching every path beneath it
func exact(p string) string {
if strings.HasSuffix(p, "/") {
internal/sink/httpchain/httpchain.go
+6 -50
View File
@@ -12,7 +12,6 @@ import (
"net/url"
"os"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
@@ -26,7 +25,6 @@ import (
"github.com/lixenwraith/logwisp/internal/sink"
"github.com/lixenwraith/logwisp/internal/tlsx"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -37,15 +35,7 @@ func init() {
}
const (
DefaultHTTPChainSinkBufferSize = 1000
DefaultHTTPChainSinkIngestPath = "/ingest"
DefaultHTTPChainSinkMaxBatchCount = 100
DefaultHTTPChainSinkMaxBatchBytes = 1024 * 1024
DefaultHTTPChainSinkFlushIntervalMS = 1000
DefaultHTTPChainSinkRequestTimeoutMS = 10000
DefaultHTTPChainSinkBackoffMinMS = 500
DefaultHTTPChainSinkBackoffMaxMS = 30000
maxResponseDrain = 64 * 1024
maxResponseDrain = 64 * 1024
)
// HTTPChainSink batches structured entries and posts NDJSON to a downstream
@@ -94,47 +84,13 @@ func NewHTTPChainSinkPlugin(
logger *log.Logger,
proxy *session.Proxy,
) (sink.Sink, error) {
opts := &config.HTTPChainSinkOptions{}
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
}
if err := lconfig.NonEmpty(opts.Host); err != nil {
return nil, fmt.Errorf("host: %w", err)
}
if err := lconfig.Port(opts.Port); err != nil {
return nil, fmt.Errorf("port: %w", err)
opts, err := config.Decode[config.HTTPChainSinkOptions]("sink", "http_chain", configMap)
if err != nil {
return nil, err
}
network, err := core.Network(opts.Host)
if err != nil {
return nil, fmt.Errorf("host: %w", err)
}
if opts.IngestPath == "" {
opts.IngestPath = DefaultHTTPChainSinkIngestPath
} else if !strings.HasPrefix(opts.IngestPath, "/") {
return nil, fmt.Errorf("ingest_path: must start with '/'")
} else if opts.IngestPath == chain.AuthPath {
return nil, fmt.Errorf("ingest_path: %s is reserved for authentication", chain.AuthPath)
}
if opts.BufferSize <= 0 {
opts.BufferSize = DefaultHTTPChainSinkBufferSize
}
if opts.MaxBatchCount <= 0 {
opts.MaxBatchCount = DefaultHTTPChainSinkMaxBatchCount
}
if opts.MaxBatchBytes <= 0 {
opts.MaxBatchBytes = DefaultHTTPChainSinkMaxBatchBytes
}
if opts.FlushIntervalMS <= 0 {
opts.FlushIntervalMS = DefaultHTTPChainSinkFlushIntervalMS
}
if opts.RequestTimeoutMS <= 0 {
opts.RequestTimeoutMS = DefaultHTTPChainSinkRequestTimeoutMS
}
if opts.BackoffMinMS <= 0 {
opts.BackoffMinMS = DefaultHTTPChainSinkBackoffMinMS
}
if opts.BackoffMaxMS < opts.BackoffMinMS {
opts.BackoffMaxMS = DefaultHTTPChainSinkBackoffMaxMS
return nil, err
}
node := opts.Node
@@ -150,7 +106,7 @@ func NewHTTPChainSinkPlugin(
if err != nil {
return nil, err
}
authPolicy, err := authz.New(opts.Auth, tlsCfg, nil, authz.RoleDialer, authz.HTTP)
authPolicy, err := authz.New(opts.Auth, tlsCfg, nil, config.Dialer, authz.HTTP)
if err != nil {
return nil, err
}
internal/sink/httpchain/httpchain_test.go
+1 -1
View File
@@ -80,7 +80,7 @@ func TestRefusedTokenLogsInAgainAndDeliversOnce(t *testing.T) {
t.Fatal(err)
}
serverTLS := &tls.Config{Certificates: []tls.Certificate{serverCert}}
listener, err := authz.New(&config.AuthOptions{Type: authz.MethodSCRAM, CredentialsFile: creds}, serverTLS, nil, authz.RoleChainListener, authz.HTTP)
listener, err := authz.New(&config.AuthOptions{Type: authz.MethodSCRAM, CredentialsFile: creds}, serverTLS, nil, config.ChainListener, authz.HTTP)
if err != nil {
t.Fatal(err)
}
internal/sink/null/null.go
+2 -2
View File
@@ -50,8 +50,8 @@ func NewNullSinkPlugin(
logger *log.Logger,
proxy *session.Proxy,
) (sink.Sink, error) {
if err := config.Scan(configMap, &config.NullSinkOptions{}); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
if _, err := config.Decode[config.NullSinkOptions]("sink", "null", configMap); err != nil {
return nil, err
}
ns := &NullSink{
id: id,
internal/sink/tcp/tcp.go
+7 -32
View File
@@ -22,7 +22,6 @@ import (
"github.com/lixenwraith/logwisp/internal/sink"
"github.com/lixenwraith/logwisp/internal/tlsx"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -32,14 +31,6 @@ func init() {
}
}
const (
DefaultTCPHost = "0.0.0.0"
DefaultTCPBufferSize = 1000
DefaultTCPClientBufferSize = 256
DefaultTCPWriteTimeoutMS = 5000
DefaultTCPKeepAlivePeriodMS = 30000
)
// TCPSink streams formatted log entries to connected TCP clients
// Concurrency model: one broadcast loop fans out into bounded per-client queues
// each connection owns a writer goroutine (drains queue) and a reader goroutine (disconnect detection)
@@ -117,31 +108,13 @@ func NewTCPSinkPlugin(
logger *log.Logger,
proxy *session.Proxy,
) (sink.Sink, error) {
opts := &config.TCPSinkOptions{
Host: DefaultTCPHost,
KeepAlive: true,
}
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
}
if err := lconfig.Port(opts.Port); err != nil {
return nil, fmt.Errorf("port: %w", err)
opts, err := config.Decode[config.TCPSinkOptions]("sink", "tcp", configMap)
if err != nil {
return nil, err
}
network, err := core.Network(opts.Host)
if err != nil {
return nil, fmt.Errorf("host: %w", err)
}
if opts.BufferSize <= 0 {
opts.BufferSize = DefaultTCPBufferSize
}
if opts.ClientBufferSize <= 0 {
opts.ClientBufferSize = DefaultTCPClientBufferSize
}
if opts.WriteTimeoutMS <= 0 {
opts.WriteTimeoutMS = DefaultTCPWriteTimeoutMS
}
if opts.KeepAlivePeriodMS <= 0 {
opts.KeepAlivePeriodMS = DefaultTCPKeepAlivePeriodMS
return nil, err
}
tlsCfg, err := tlsx.Server(opts.TLS, opts.Host)
if err != nil {
@@ -151,7 +124,7 @@ func NewTCPSinkPlugin(
if err != nil {
return nil, err
}
authPolicy, err := authz.New(opts.Auth, tlsCfg, aclPolicy, authz.RoleListener, authz.TCP)
authPolicy, err := authz.New(opts.Auth, tlsCfg, aclPolicy, config.Listener, authz.TCP)
if err != nil {
return nil, err
}
@@ -217,6 +190,8 @@ func (t *TCPSink) listen() (net.Listener, error) {
Enable: true,
Idle: time.Duration(t.config.KeepAlivePeriodMS) * time.Millisecond,
}
} else {
lc.KeepAlive = -1 // zero would be Go's default, keep-alive every 15 s
}
ln, err := core.Listen(context.Background(), &lc, t.network, t.addr)
if err != nil {
internal/sink/tcpchain/tcpchain.go
+7 -42
View File
@@ -23,7 +23,6 @@ import (
"github.com/lixenwraith/logwisp/internal/sink"
"github.com/lixenwraith/logwisp/internal/tlsx"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -33,15 +32,6 @@ func init() {
}
}
const (
DefaultChainSinkBufferSize = 1000
DefaultChainSinkDialTimeoutMS = 5000
DefaultChainSinkWriteTimeoutMS = 5000
DefaultChainSinkBackoffMinMS = 500
DefaultChainSinkBackoffMaxMS = 30000
DefaultChainSinkKeepAlivePeriodMS = 30000
)
// TCPChainSink forwards structured entries to a downstream tcp_chain source
type TCPChainSink struct {
id string
@@ -89,40 +79,13 @@ func NewTCPChainSinkPlugin(
logger *log.Logger,
proxy *session.Proxy,
) (sink.Sink, error) {
opts := &config.TCPChainSinkOptions{
KeepAlive: true,
}
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
}
if err := lconfig.NonEmpty(opts.Host); err != nil {
return nil, fmt.Errorf("host: %w", err)
}
if err := lconfig.Port(opts.Port); err != nil {
return nil, fmt.Errorf("port: %w", err)
opts, err := config.Decode[config.TCPChainSinkOptions]("sink", "tcp_chain", configMap)
if err != nil {
return nil, err
}
network, err := core.Network(opts.Host)
if err != nil {
return nil, fmt.Errorf("host: %w", err)
}
if opts.BufferSize <= 0 {
opts.BufferSize = DefaultChainSinkBufferSize
}
if opts.DialTimeoutMS <= 0 {
opts.DialTimeoutMS = DefaultChainSinkDialTimeoutMS
}
if opts.WriteTimeoutMS <= 0 {
opts.WriteTimeoutMS = DefaultChainSinkWriteTimeoutMS
}
if opts.BackoffMinMS <= 0 {
opts.BackoffMinMS = DefaultChainSinkBackoffMinMS
}
if opts.BackoffMaxMS < opts.BackoffMinMS {
opts.BackoffMaxMS = DefaultChainSinkBackoffMaxMS
}
if opts.KeepAlivePeriodMS <= 0 {
opts.KeepAlivePeriodMS = DefaultChainSinkKeepAlivePeriodMS
return nil, err
}
node := opts.Node
@@ -138,7 +101,7 @@ func NewTCPChainSinkPlugin(
if err != nil {
return nil, err
}
authPolicy, err := authz.New(opts.Auth, tlsCfg, nil, authz.RoleDialer, authz.TCP)
authPolicy, err := authz.New(opts.Auth, tlsCfg, nil, config.Dialer, authz.TCP)
if err != nil {
return nil, err
}
@@ -378,6 +341,8 @@ func (t *TCPChainSink) connect(ctx context.Context) error {
Enable: true,
Idle: time.Duration(t.config.KeepAlivePeriodMS) * time.Millisecond,
}
} else {
nd.KeepAlive = -1 // zero would be Go's default, keep-alive every 15 s
}
var conn net.Conn
internal/source/console/console.go
+3 -22
View File
@@ -25,13 +25,6 @@ func init() {
panic(fmt.Sprintf("failed to register console source: %v", err))
}
// Console stdin can only have one reader
if err := plugin.SetSourceMetadata("console", &plugin.PluginMetadata{
Capabilities: []core.Capability{core.CapSessionAware, core.CapSingleInstance},
MaxInstances: 1,
}); err != nil {
panic(fmt.Sprintf("failed to set console source metadata: %v", err))
}
}
// One reader serves the process: after a reload the new source continues
@@ -114,10 +107,6 @@ type ConsoleSource struct {
ended atomic.Bool // standard input reached its end
}
const (
DefaultConsoleSourceBufferSize = 1000
)
// NewConsoleSourcePlugin creates a console source through plugin factory
func NewConsoleSourcePlugin(
id string,
@@ -125,19 +114,11 @@ func NewConsoleSourcePlugin(
logger *log.Logger,
proxy *session.Proxy,
) (source.Source, error) {
opts := &config.ConsoleSourceOptions{}
// Scan config map
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
opts, err := config.Decode[config.ConsoleSourceOptions]("source", "console", configMap)
if err != nil {
return nil, err
}
// Validate and apply defaults
if opts.BufferSize <= 0 {
opts.BufferSize = DefaultConsoleSourceBufferSize
}
// Create and return plugin instance
cs := &ConsoleSource{
id: id,
proxy: proxy,
internal/source/file/file.go
+3 -33
View File
@@ -19,7 +19,6 @@ import (
"github.com/lixenwraith/logwisp/internal/session"
"github.com/lixenwraith/logwisp/internal/source"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -64,13 +63,6 @@ type FileSource struct {
lastEntryTime atomic.Value // time.Time
}
const (
DefaultFileSourcePattern = "*"
DefaultFileSourceCheckIntervalMS = 100
MinFileSourceCheckIntervalMS = 10
DefaultFileSourceFrom = "end"
)
// NewFileSourcePlugin creates a file source through plugin factory
func NewFileSourcePlugin(
id string,
@@ -78,33 +70,11 @@ func NewFileSourcePlugin(
logger *log.Logger,
proxy *session.Proxy,
) (source.Source, error) {
opts := &config.FileSourceOptions{}
// Use lconfig to scan map into struct (overriding defaults)
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
opts, err := config.Decode[config.FileSourceOptions]("source", "file", configMap)
if err != nil {
return nil, err
}
// Validate and apply defaults
if err := lconfig.NonEmpty(opts.Directory); err != nil {
return nil, fmt.Errorf("directory: %w", err)
}
if opts.Pattern == "" {
opts.Pattern = DefaultFileSourcePattern
}
if opts.CheckIntervalMS <= 0 {
opts.CheckIntervalMS = DefaultFileSourceCheckIntervalMS
} else if opts.CheckIntervalMS < MinFileSourceCheckIntervalMS {
return nil, fmt.Errorf("check_interval_ms: must be >= %d", MinFileSourceCheckIntervalMS)
}
if opts.From == "" {
opts.From = DefaultFileSourceFrom
} else if err := lconfig.OneOf("start", "end")(opts.From); err != nil {
return nil, fmt.Errorf("from: %w", err)
}
// Create and return plugin instance
fs := &FileSource{
id: id,
proxy: proxy,
internal/source/httpchain/httpchain.go
+7 -35
View File
@@ -10,7 +10,6 @@ import (
"net"
"net/http"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
@@ -25,7 +24,6 @@ import (
"github.com/lixenwraith/logwisp/internal/source"
"github.com/lixenwraith/logwisp/internal/tlsx"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -36,12 +34,8 @@ func init() {
}
const (
DefaultHTTPChainSourceBufferSize = 1000
DefaultHTTPChainSourceIngestPath = "/ingest"
DefaultHTTPChainSourceMaxBodyBytes = 8 * 1024 * 1024
DefaultHTTPChainSourceReadTimeoutMS = 30000
HTTPChainReadHeaderTimeout = 10 * time.Second
HTTPChainServerShutdownTimeout = 2 * time.Second
HTTPChainReadHeaderTimeout = 10 * time.Second
HTTPChainServerShutdownTimeout = 2 * time.Second
)
// HTTPChainSource accepts NDJSON batches from upstream http_chain sinks
@@ -84,35 +78,13 @@ func NewHTTPChainSourcePlugin(
logger *log.Logger,
proxy *session.Proxy,
) (source.Source, error) {
opts := &config.HTTPChainSourceOptions{
Host: "0.0.0.0",
TrustNode: true,
}
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
}
if err := lconfig.Port(opts.Port); err != nil {
return nil, fmt.Errorf("port: %w", err)
opts, err := config.Decode[config.HTTPChainSourceOptions]("source", "http_chain", configMap)
if err != nil {
return nil, err
}
network, err := core.Network(opts.Host)
if err != nil {
return nil, fmt.Errorf("host: %w", err)
}
if opts.IngestPath == "" {
opts.IngestPath = DefaultHTTPChainSourceIngestPath
} else if !strings.HasPrefix(opts.IngestPath, "/") {
return nil, fmt.Errorf("ingest_path: must start with '/'")
} else if opts.IngestPath == chain.AuthPath {
return nil, fmt.Errorf("ingest_path: %s is reserved for authentication", chain.AuthPath)
}
if opts.BufferSize <= 0 {
opts.BufferSize = DefaultHTTPChainSourceBufferSize
}
if opts.MaxBodyBytes <= 0 {
opts.MaxBodyBytes = DefaultHTTPChainSourceMaxBodyBytes
}
if opts.ReadTimeoutMS <= 0 {
opts.ReadTimeoutMS = DefaultHTTPChainSourceReadTimeoutMS
return nil, err
}
tlsCfg, err := tlsx.Server(opts.TLS, opts.Host)
if err != nil {
@@ -122,7 +94,7 @@ func NewHTTPChainSourcePlugin(
if err != nil {
return nil, err
}
authPolicy, err := authz.New(opts.Auth, tlsCfg, aclPolicy, authz.RoleChainListener, authz.HTTP)
authPolicy, err := authz.New(opts.Auth, tlsCfg, aclPolicy, config.ChainListener, authz.HTTP)
if err != nil {
return nil, err
}
internal/source/null/null.go
+2 -2
View File
@@ -48,8 +48,8 @@ func NewNullSourcePlugin(
logger *log.Logger,
proxy *session.Proxy,
) (source.Source, error) {
if err := config.Scan(configMap, &config.NullSourceOptions{}); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
if _, err := config.Decode[config.NullSourceOptions]("source", "null", configMap); err != nil {
return nil, err
}
ns := &NullSource{
id: id,
internal/source/random/random.go
+3 -42
View File
@@ -14,7 +14,6 @@ import (
"github.com/lixenwraith/logwisp/internal/session"
"github.com/lixenwraith/logwisp/internal/source"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -53,12 +52,6 @@ type RandomSource struct {
lastEntryTime atomic.Value // time.Time
}
const (
DefaultRandomSourceIntervalMS = 500
DefaultRandomSourceFormat = "txt"
DefaultRandomSourceLength = 20
)
// NewRandomSourcePlugin creates a random source through plugin factory
func NewRandomSourcePlugin(
id string,
@@ -67,41 +60,9 @@ func NewRandomSourcePlugin(
proxy *session.Proxy,
) (source.Source, error) {
// Step 1: Create empty config struct with defaults
opts := &config.RandomSourceOptions{
IntervalMS: 500,
JitterMS: 0,
Format: "txt",
Length: 20,
Special: false,
}
// Scan config map
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
}
// Defaults
if opts.IntervalMS <= 0 {
opts.IntervalMS = DefaultRandomSourceIntervalMS
}
if opts.Format == "" {
opts.Format = DefaultRandomSourceFormat
}
if opts.Length <= 0 {
opts.Length = DefaultRandomSourceLength
}
// Validate
if opts.JitterMS < 0 {
return nil, fmt.Errorf("jitter_ms cannot be negative")
}
if opts.JitterMS > opts.IntervalMS {
opts.JitterMS = opts.IntervalMS
}
validateFormat := lconfig.OneOf("raw", "txt", "json")
if err := validateFormat(opts.Format); err != nil {
return nil, fmt.Errorf("format: %w", err)
opts, err := config.Decode[config.RandomSourceOptions]("source", "random", configMap)
if err != nil {
return nil, err
}
rs := &RandomSource{
internal/source/tcpchain/tcpchain.go
+5 -23
View File
@@ -23,7 +23,6 @@ import (
"github.com/lixenwraith/logwisp/internal/source"
"github.com/lixenwraith/logwisp/internal/tlsx"
lconfig "github.com/lixenwraith/config"
"github.com/lixenwraith/log"
)
@@ -33,11 +32,6 @@ func init() {
}
}
const (
DefaultChainSourceBufferSize = 1000
DefaultChainSourceHelloTimeoutMS = 10000
)
// TCPChainSource accepts connections from upstream tcp_chain sinks and ingests NDJSON entries
type TCPChainSource struct {
id string
@@ -79,25 +73,13 @@ func NewTCPChainSourcePlugin(
logger *log.Logger,
proxy *session.Proxy,
) (source.Source, error) {
opts := &config.TCPChainSourceOptions{
Host: "0.0.0.0",
TrustNode: true,
}
if err := config.Scan(configMap, opts); err != nil {
return nil, fmt.Errorf("failed to parse config: %w", err)
}
if err := lconfig.Port(opts.Port); err != nil {
return nil, fmt.Errorf("port: %w", err)
opts, err := config.Decode[config.TCPChainSourceOptions]("source", "tcp_chain", configMap)
if err != nil {
return nil, err
}
network, err := core.Network(opts.Host)
if err != nil {
return nil, fmt.Errorf("host: %w", err)
}
if opts.BufferSize <= 0 {
opts.BufferSize = DefaultChainSourceBufferSize
}
if opts.HelloTimeoutMS <= 0 {
opts.HelloTimeoutMS = DefaultChainSourceHelloTimeoutMS
return nil, err
}
tlsCfg, err := tlsx.Server(opts.TLS, opts.Host)
if err != nil {
@@ -107,7 +89,7 @@ func NewTCPChainSourcePlugin(
if err != nil {
return nil, err
}
authPolicy, err := authz.New(opts.Auth, tlsCfg, aclPolicy, authz.RoleChainListener, authz.TCP)
authPolicy, err := authz.New(opts.Auth, tlsCfg, aclPolicy, config.ChainListener, authz.TCP)
if err != nil {
return nil, err
}
internal/tlsx/pki.go
+6 -10
View File
@@ -23,6 +23,8 @@ import (
"sync"
"time"
"unicode"
"github.com/lixenwraith/logwisp/internal/config"
)
// Certificates are ECDSA P-256, for lw tls and for listeners without files.
@@ -31,7 +33,6 @@ import (
const (
CAValidity = 10 * 365 * 24 * time.Hour
LeafValidity = 397 * 24 * time.Hour
pinPrefix = "sha256//"
)
// processKey is the key of every generated listener certificate: reloads
@@ -184,21 +185,16 @@ func pinOfKey(k *ecdsa.PrivateKey) string {
func pin(spki []byte) string {
h := sha256.Sum256(spki)
return pinPrefix + base64.StdEncoding.EncodeToString(h[:])
return config.PinPrefix + base64.StdEncoding.EncodeToString(h[:])
}
// verifyPins replaces chain verification: the server's key must hash to one
// of the ';'-separated pins. Its name and validity are not checked. Go skips
// it on a resumed session, which Client therefore disables.
func verifyPins(s string) (func([][]byte, [][]*x509.Certificate) error, error) {
var pins [][]byte
for p := range strings.SplitSeq(s, ";") {
b64, ok := strings.CutPrefix(strings.TrimSpace(p), pinPrefix)
h, err := base64.StdEncoding.DecodeString(b64)
if !ok || err != nil || len(h) != sha256.Size {
return nil, fmt.Errorf("tls: pin_sha256 %q: want %sBASE64 of a SHA-256, ';' between several", p, pinPrefix)
}
pins = append(pins, h)
pins, err := config.ParsePins(s)
if err != nil {
return nil, err
}
return func(raw [][]byte, _ [][]*x509.Certificate) error {
if len(raw) == 0 {
internal/tlsx/tlsx.go
+5 -26
View File
@@ -34,24 +34,12 @@ func Server(o *config.TLSOptions, host string) (*tls.Config, error) {
if o == nil || !o.Enabled {
return nil, nil
}
files, issuer := o.CertFile != "" || o.KeyFile != "", o.IssuerCertFile != "" || o.IssuerKeyFile != ""
switch {
case o.CAFile != "" || o.ServerName != "" || o.InsecureSkipVerify || o.PinSHA256 != "":
return nil, fmt.Errorf("tls: ca_file, server_name, insecure_skip_verify and pin_sha256 apply to dialers")
case files && (issuer || o.SelfSigned) || issuer && o.SelfSigned:
return nil, fmt.Errorf("tls: set one of cert_file and key_file, self_signed, or issuer_cert_file and issuer_key_file")
case len(o.Hosts) > 0 && !issuer && !o.SelfSigned:
return nil, fmt.Errorf("tls: hosts applies to self_signed and issuer certificates")
case files && (o.CertFile == "" || o.KeyFile == ""):
return nil, fmt.Errorf("tls: cert_file and key_file must be set together")
case issuer && (o.IssuerCertFile == "" || o.IssuerKeyFile == ""):
return nil, fmt.Errorf("tls: issuer_cert_file and issuer_key_file must be set together")
case !files && !issuer && !o.SelfSigned:
return nil, fmt.Errorf("tls: listeners need cert_file and key_file, self_signed, or issuer_cert_file and issuer_key_file")
if err := o.Check(true); err != nil {
return nil, err
}
var cert tls.Certificate
var err error
if files {
if o.CertFile != "" {
cert, err = tls.LoadX509KeyPair(o.CertFile, o.KeyFile)
} else {
cert, err = generated(o.IssuerCertFile, o.IssuerKeyFile, host, o.Hosts)
@@ -68,9 +56,6 @@ func Server(o *config.TLSOptions, host string) (*tls.Config, error) {
MinVersion: mv,
}
if o.ClientAuth {
if o.ClientCAFile == "" {
return nil, fmt.Errorf("tls: client_auth requires client_ca_file")
}
pool, err := loadPool(o.ClientCAFile)
if err != nil {
return nil, err
@@ -90,11 +75,8 @@ func Client(o *config.TLSOptions, host string) (*tls.Config, error) {
if o == nil || !o.Enabled {
return nil, nil
}
switch {
case o.SelfSigned || o.IssuerCertFile != "" || o.IssuerKeyFile != "" || len(o.Hosts) > 0 || o.ClientAuth || o.ClientCAFile != "":
return nil, fmt.Errorf("tls: self_signed, issuer_cert_file, issuer_key_file, hosts, client_auth and client_ca_file apply to listeners")
case o.PinSHA256 != "" && (o.CAFile != "" || o.InsecureSkipVerify):
return nil, fmt.Errorf("tls: pin_sha256 replaces ca_file and insecure_skip_verify: set one")
if err := o.Check(false); err != nil {
return nil, err
}
mv, err := minVersion(o.MinVersion)
if err != nil {
@@ -123,9 +105,6 @@ func Client(o *config.TLSOptions, host string) (*tls.Config, error) {
}
cfg.RootCAs = pool
}
if (o.CertFile == "") != (o.KeyFile == "") {
return nil, fmt.Errorf("tls: cert_file and key_file must be set together")
}
if o.CertFile != "" {
cert, err := tls.LoadX509KeyPair(o.CertFile, o.KeyFile)
if err != nil {