diff --git a/.env.example b/.env.example index f379fe5..2ea09ff 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,14 @@ KEYWARDEN_DEBUG=True KEYWARDEN_ALLOWED_HOSTS=* KEYWARDEN_TRUSTED_ORIGINS=https://reverse.proxy.domain.xyz,https://127.0.0.1 KEYWARDEN_DOMAIN=https://example.domain.xyz +KEYWARDEN_SHELL_REMOTE_COMMAND=/bin/bash +# Set to 1 to include SSH probe/certificate diagnostics in shell failures. +KEYWARDEN_SHELL_DEBUG=0 +KEYWARDEN_SHELL_PRESERVE_TMP=0 +KEYWARDEN_SHELL_DIAG_TIMEOUT_SECONDS=10 +# Optional legacy global fallback token for agents. +# Preferred model is per-agent token issued during /agent/enroll. +# KEYWARDEN_AGENT_API_TOKEN= # Database KEYWARDEN_POSTGRES_DB=keywarden diff --git a/agent/LOGGING_PACKAGES.md b/agent/LOGGING_PACKAGES.md new file mode 100644 index 0000000..909d77a --- /dev/null +++ b/agent/LOGGING_PACKAGES.md @@ -0,0 +1,35 @@ +# Logging Package Options + +This agent already uses a library approach for journald (`github.com/coreos/go-systemd/v22/sdjournal`) and does not shell out to `journalctl` in normal collection paths. + +## Current (keep) + +- `github.com/coreos/go-systemd/v22/sdjournal` + - Direct journal access from Go. + - Cursor-based reads map well to incremental shipping. +- `github.com/coreos/go-systemd/v22/dbus` (candidate add-on) + - Read systemd unit state/metadata (`ListUnitsByNames`, unit properties) for richer service context. + +## Good external collectors (if we want hardened third-party gathering) + +- Fluent Bit + - `systemd` input plugin for journald, plus mature filtering/parsing/routing outputs. + - Strong choice when we want a small footprint daemon dedicated to log forwarding. +- Vector + - `journald` source + transforms (VRL) for compact line formatting before shipping. + - Good if we want expressive local transforms and multiple downstream sinks. +- OpenTelemetry Collector distributions with journald receiver support (for example Splunk OTel distro) + - Useful if we want logs + metrics + traces through one pipeline. + +## Parser packages (when we need deeper line parsing) + +- `github.com/influxdata/go-syslog/v3` + - RFC3164/RFC5424 syslog parsing for file or stream logs. +- `github.com/elastic/go-grok` + - Pattern-based parsing when message formats vary by service. + +## Practical direction + +1. Keep `go-systemd` for now and add optional `dbus` lookups for service status enrichment. +2. If we need host-agnostic, independently hardened collection, run Fluent Bit or Vector and ingest normalized JSON lines in the agent/server. +3. Reserve heavier parsing packages (`go-syslog`, `go-grok`) for specific source formats that require structured extraction beyond current compact line ingestion. diff --git a/agent/README.md b/agent/README.md index 24219e9..7f59f63 100644 --- a/agent/README.md +++ b/agent/README.md @@ -16,12 +16,28 @@ go build -o keywarden-agent ./cmd/keywarden-agent ./keywarden-agent -config /etc/keywarden/agent.json -server-url https://keywarden.example.com -enroll-token ``` +To rotate/re-enroll an existing agent identity, add `-force-enroll` with a fresh token. + You can also pass `KEYWARDEN_SERVER_URL` and `KEYWARDEN_ENROLL_TOKEN` as environment variables. ## Config On first boot, the agent will create a config file if it does not exist. Only `server_url` is required for bootstrapping. +Do not prefill `server_id` or `agent_api_token` with placeholders; those are written by enrollment. If the Keywarden server uses a private TLS CA, set `server_ca_path` (or `KEYWARDEN_SERVER_CA_PATH`) to the CA PEM file so the agent can verify the server certificate. +`agent_api_token` is issued by `/agent/enroll` and persisted into `agent.json` automatically. +Use the raw token value (no `Bearer ` prefix). If `KEYWARDEN_AGENT_API_TOKEN` env var is set, it overrides `agent_api_token` in `agent.json`. +When no `log_sources` are configured, the agent defaults to journald collection via `go-systemd/sdjournal` and backfills from the start of the current boot on first sync. +If journald is unavailable, the agent falls back to common system log files (`/var/log/auth.log`, `/var/log/secure`, `/var/log/syslog`, `/var/log/messages`). + +`log_sources` can be defined locally for service/file monitors. These are used as: +- fallback if central `/agent/servers/{id}/log-config` cannot be fetched +- additive sources when central config defines one or more explicit sources + +If central `log-config` returns an empty list, the agent uses default journald +collection from the current boot and ignores local `log_sources` for that run. See `config.example.json`. + +For package alternatives and external collector options, see `LOGGING_PACKAGES.md`. diff --git a/agent/cmd/keywarden-agent/main.go b/agent/cmd/keywarden-agent/main.go index 4f2f813..7438a9c 100644 --- a/agent/cmd/keywarden-agent/main.go +++ b/agent/cmd/keywarden-agent/main.go @@ -4,15 +4,20 @@ import ( "context" "crypto/rand" "crypto/rsa" + "crypto/sha256" "crypto/x509" "crypto/x509/pkix" + "encoding/hex" "encoding/json" "encoding/pem" + "errors" "flag" "fmt" "log" "os" "os/signal" + "sort" + "strings" "syscall" "time" @@ -23,10 +28,13 @@ import ( "keywarden/agent/internal/version" ) +var lastNoEventsNotice time.Time + func main() { configPath := flag.String("config", config.DefaultConfigPath, "Path to agent config JSON") serverURL := flag.String("server-url", "", "Keywarden server URL (first boot)") enrollToken := flag.String("enroll-token", "", "Enrollment token (first boot)") + forceEnroll := flag.Bool("force-enroll", false, "Force re-enrollment even if already bootstrapped") showVersion := flag.Bool("version", false, "Print version and exit") flag.Parse() @@ -43,7 +51,7 @@ func main() { log.Fatalf("state dir error: %v", err) } - if err := bootstrapIfNeeded(cfg, *configPath, pickEnrollToken(*enrollToken)); err != nil { + if err := bootstrapIfNeeded(cfg, *configPath, pickEnrollToken(*enrollToken), *forceEnroll); err != nil { log.Fatalf("bootstrap error: %v", err) } @@ -80,20 +88,30 @@ func runOnce(ctx context.Context, apiClient *client.Client, cfg *config.Config) log.Printf("host update deferred; will retry: %v", err) } else { log.Printf("host update error: %v", err) + logAgentUnauthorizedHint(err) } } if err := apiClient.SyncAccounts(ctx, cfg); err != nil { log.Printf("sync accounts error: %v", err) + logAgentUnauthorizedHint(err) } if err := shipLogs(ctx, apiClient, cfg); err != nil { if client.IsRetriable(err) { log.Printf("log shipping deferred; will retry: %v", err) } else { log.Printf("log shipping error: %v", err) + logAgentUnauthorizedHint(err) } } } +func logAgentUnauthorizedHint(err error) { + var statusErr *client.HTTPStatusError + if errors.As(err, &statusErr) && statusErr.StatusCode == 401 { + log.Printf("agent auth failed (401): ensure agent_api_token matches the enrollment-issued token for this server (raw token only; no 'Bearer ' prefix)") + } +} + func ensureDirs(cfg *config.Config) error { if err := os.MkdirAll(cfg.StateDir, 0o700); err != nil { return err @@ -114,16 +132,56 @@ func shipLogs(ctx context.Context, apiClient *client.Client, cfg *config.Config) return err } - cursor, err := logs.ReadCursor(cfg.LogCursorPath()) + state, err := logs.ReadState(cfg.LogStatePath()) + if err != nil { + return err + } + localSources := toLogSourcesFromConfig(cfg.EnabledLogSources()) + sourceConfig, err := apiClient.FetchLogConfig(ctx, cfg.ServerID) + var sources []logs.SourceConfig + if err != nil { + if len(localSources) == 0 { + return err + } + log.Printf("log config fetch failed; using local log_sources fallback: %v", err) + sources = localSources + } else { + sources = toLogSources(sourceConfig) + if len(sources) > 0 && len(localSources) > 0 { + sources = append(sources, localSources...) + } + } + state, err = reconcileLogState(cfg, state, sources) if err != nil { return err } collector := logs.NewCollector() - events, nextCursor, err := collector.Collect(ctx, cursor, cfg.LogBatchSize) + events, nextState, err := collector.Collect(ctx, cfg.LogBatchSize, sources, state) if err != nil { - return err + if len(sources) > 0 { + return err + } + fallbackSources := defaultFileFallbackSources() + log.Printf("default journald collection failed; falling back to file monitors: %v", err) + events, nextState, err = collector.Collect(ctx, cfg.LogBatchSize, fallbackSources, state) + if err != nil { + return err + } + } + if len(events) == 0 && len(sources) == 0 { + fallbackSources := defaultFileFallbackSources() + log.Printf("default journald yielded no events; trying file monitors") + fileEvents, fileState, fileErr := collector.Collect(ctx, cfg.LogBatchSize, fallbackSources, nextState) + if fileErr != nil { + return fileErr + } + events = fileEvents + nextState = fileState } if len(events) == 0 { + if maybeLogNoEventsNotice(len(sources)) && len(sources) == 0 { + log.Printf("log shipping file fallback probe: %s", describeFileSourceAvailability(defaultFileFallbackSources())) + } return nil } payload, err := json.Marshal(events) @@ -136,7 +194,7 @@ func shipLogs(ctx context.Context, apiClient *client.Client, cfg *config.Config) } return err } - if err := logs.WriteCursor(cfg.LogCursorPath(), nextCursor); err != nil { + if err := logs.WriteState(cfg.LogStatePath(), nextState); err != nil { return err } return nil @@ -172,13 +230,22 @@ func pickEnrollToken(flagValue string) string { return os.Getenv("KEYWARDEN_ENROLL_TOKEN") } -func bootstrapIfNeeded(cfg *config.Config, configPath string, enrollToken string) error { - if cfg.ServerID != "" && fileExists(cfg.ClientCertPath()) && fileExists(cfg.CACertPath()) { +func bootstrapIfNeeded(cfg *config.Config, configPath string, enrollToken string, forceEnroll bool) error { + previousServerID := strings.TrimSpace(cfg.ServerID) + if !forceEnroll && cfg.ServerID != "" && fileExists(cfg.ClientCertPath()) && fileExists(cfg.CACertPath()) { + if strings.TrimSpace(enrollToken) != "" { + log.Printf("enrollment token provided but bootstrap is already complete; skipping enrollment (use -force-enroll to rotate identity)") + } return nil } if enrollToken == "" { return fmt.Errorf("missing enrollment token; set KEYWARDEN_ENROLL_TOKEN or -enroll-token") } + if forceEnroll { + log.Printf("force re-enrollment requested; replacing existing server identity") + cfg.ServerID = "" + cfg.AgentAPIToken = "" + } keyPath := cfg.ClientKeyPath() if !fileExists(keyPath) { if err := generateKey(keyPath); err != nil { @@ -208,12 +275,156 @@ func bootstrapIfNeeded(cfg *config.Config, configPath string, enrollToken string return err } cfg.ServerID = resp.ServerID + if token := strings.TrimSpace(resp.AgentToken); token != "" { + cfg.AgentAPIToken = token + } + if forceEnroll || (previousServerID != "" && previousServerID != cfg.ServerID) { + if err := resetLogState(cfg); err != nil { + return fmt.Errorf("reset log state: %w", err) + } + } if err := config.Save(configPath, cfg); err != nil { return err } return nil } +func resetLogState(cfg *config.Config) error { + if err := os.Remove(cfg.LogCursorPath()); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + if err := os.Remove(cfg.LogOffsetsPath()); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + if err := os.Remove(cfg.LogStatePath()); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + if err := os.Remove(logSourceStatePath(cfg)); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + if err := os.RemoveAll(cfg.LogSpoolDir()); err != nil { + return err + } + return os.MkdirAll(cfg.LogSpoolDir(), 0o700) +} + +func defaultFileFallbackSources() []logs.SourceConfig { + return []logs.SourceConfig{ + {Kind: "file", Name: "/var/log/auth.log", FilePath: "/var/log/auth.log", Category: "auth", EventType: "file.line", Parser: "syslog"}, + {Kind: "file", Name: "/var/log/secure", FilePath: "/var/log/secure", Category: "auth", EventType: "file.line", Parser: "syslog"}, + {Kind: "file", Name: "/var/log/syslog", FilePath: "/var/log/syslog", Category: "system", EventType: "file.line", Parser: "syslog"}, + {Kind: "file", Name: "/var/log/messages", FilePath: "/var/log/messages", Category: "system", EventType: "file.line", Parser: "syslog"}, + } +} + +func maybeLogNoEventsNotice(sourceCount int) bool { + now := time.Now() + if !lastNoEventsNotice.IsZero() && now.Sub(lastNoEventsNotice) < time.Minute { + return false + } + mode := "default journald (current boot)" + if sourceCount > 0 { + mode = "configured sources" + } + log.Printf("log shipping: no events collected (%s)", mode) + lastNoEventsNotice = now + return true +} + +func describeFileSourceAvailability(sources []logs.SourceConfig) string { + items := make([]string, 0, len(sources)) + for _, source := range sources { + path := strings.TrimSpace(source.FilePath) + if path == "" { + path = strings.TrimSpace(source.Name) + } + if path == "" { + continue + } + item := path + ": " + info, err := os.Stat(path) + if err != nil { + if os.IsNotExist(err) { + items = append(items, item+"missing") + continue + } + items = append(items, item+err.Error()) + continue + } + if info.IsDir() { + items = append(items, item+"is directory") + continue + } + file, openErr := os.Open(path) + if openErr != nil { + items = append(items, item+openErr.Error()) + continue + } + _ = file.Close() + items = append(items, fmt.Sprintf("%d bytes readable", info.Size())) + } + if len(items) == 0 { + return "no file sources" + } + return strings.Join(items, "; ") +} + +func reconcileLogState( + cfg *config.Config, + state map[string]logs.SourceState, + sources []logs.SourceConfig, +) (map[string]logs.SourceState, error) { + statePath := logSourceStatePath(cfg) + previous, err := os.ReadFile(statePath) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return state, err + } + previousSig := strings.TrimSpace(string(previous)) + currentSig := logSourceSignature(sources) + needsReset := false + if previousSig == "" { + if len(state) > 0 { + log.Printf("log state signature missing; resetting per-source state for fresh collection") + needsReset = true + } + } else if previousSig != currentSig { + log.Printf("log sources changed; resetting per-source state for fresh collection") + needsReset = true + } + if err := os.WriteFile(statePath, []byte(currentSig+"\n"), 0o600); err != nil { + return state, err + } + if needsReset { + return map[string]logs.SourceState{}, nil + } + return state, nil +} + +func logSourceStatePath(cfg *config.Config) string { + return cfg.StateDir + "/log.sources.sha256" +} + +func logSourceSignature(sources []logs.SourceConfig) string { + parts := make([]string, 0, len(sources)) + for _, source := range sources { + parts = append(parts, strings.Join([]string{ + strings.TrimSpace(source.SourceID), + strings.ToLower(strings.TrimSpace(source.Kind)), + strings.TrimSpace(source.Name), + strings.TrimSpace(source.ServiceUnit), + strings.TrimSpace(source.FilePath), + strings.ToLower(strings.TrimSpace(source.Parser)), + strings.TrimSpace(source.Category), + strings.TrimSpace(source.EventType), + encodeMatchMap(source.IncludeMatches), + encodeMatchMap(source.ExcludeMatches), + }, "\x1f")) + } + sort.Strings(parts) + sum := sha256.Sum256([]byte(strings.Join(parts, "\x1e"))) + return hex.EncodeToString(sum[:]) +} + func retry(ctx context.Context, delays []time.Duration, fn func() error) error { var lastErr error for attempt := 0; attempt <= len(delays); attempt++ { @@ -276,3 +487,59 @@ func fileExists(path string) bool { } return !info.IsDir() } + +func toLogSources(in []client.LogSourceConfig) []logs.SourceConfig { + out := make([]logs.SourceConfig, 0, len(in)) + for _, source := range in { + out = append(out, logs.SourceConfig{ + SourceID: source.SourceID, + Kind: source.Kind, + Name: source.Name, + ServiceUnit: source.ServiceUnit, + FilePath: source.FilePath, + Parser: source.Parser, + IncludeMatches: source.IncludeMatches, + ExcludeMatches: source.ExcludeMatches, + Category: source.Category, + EventType: source.EventType, + }) + } + return out +} + +func toLogSourcesFromConfig(in []config.LogSource) []logs.SourceConfig { + out := make([]logs.SourceConfig, 0, len(in)) + for _, source := range in { + out = append(out, logs.SourceConfig{ + SourceID: source.SourceID, + Kind: source.Kind, + Name: source.Name, + ServiceUnit: source.ServiceUnit, + FilePath: source.FilePath, + Parser: source.Parser, + IncludeMatches: source.IncludeMatches, + ExcludeMatches: source.ExcludeMatches, + Category: source.Category, + EventType: source.EventType, + }) + } + return out +} + +func encodeMatchMap(in map[string][]string) string { + if len(in) == 0 { + return "" + } + keys := make([]string, 0, len(in)) + for key := range in { + keys = append(keys, key) + } + sort.Strings(keys) + parts := make([]string, 0, len(keys)) + for _, key := range keys { + values := append([]string{}, in[key]...) + sort.Strings(values) + parts = append(parts, key+"="+strings.Join(values, ",")) + } + return strings.Join(parts, ";") +} diff --git a/agent/config.example.json b/agent/config.example.json index f499b6c..affeee5 100644 --- a/agent/config.example.json +++ b/agent/config.example.json @@ -1,9 +1,59 @@ { - "server_url": "https://keywarden.dev.ntbx.io/api/v1", - "server_id": "4", - "server_ca_path": "", + "server_url": "https://keywarden.example.com/api/v1", + "server_id": "6", + "agent_api_token": "HV8MnbysmINShu9wkXzzNrQE5NyRlPJHfP2aU6SX4HERrlac8EQpoMEu02FX1TYv", "sync_interval_seconds": 5, - "log_batch_size": 500, + "log_batch_size": 2000, + "log_sources": [ + { + "source_id": "journal-all", + "kind": "journal", + "name": "All Journal", + "category": "system", + "event_type": "system", + "parser": "none", + "include_matches": {}, + "exclude_matches": {} + }, + { + "source_id": "journal-kernel", + "kind": "journal", + "name": "Kernel Journal", + "category": "system", + "event_type": "kernel", + "parser": "none", + "include_matches": { + "_TRANSPORT": ["kernel"] + } + }, + { + "source_id": "service-ssh", + "kind": "service", + "name": "OpenSSH", + "service_unit": "ssh.service", + "category": "access", + "event_type": "ssh", + "parser": "none" + }, + { + "source_id": "file-auth", + "kind": "file", + "name": "Auth Log", + "file_path": "/var/log/auth.log", + "category": "auth", + "event_type": "file.line", + "parser": "syslog" + }, + { + "source_id": "file-nginx-access", + "kind": "file", + "name": "Nginx Access", + "file_path": "/var/log/nginx/access.log", + "category": "access", + "event_type": "http.access", + "parser": "nginx_access" + } + ], "state_dir": "/var/lib/keywarden-agent", "account_policy": { "username_template": "{{username}}_{{user_id}}", diff --git a/agent/config.json b/agent/config.json new file mode 100644 index 0000000..68dec6c --- /dev/null +++ b/agent/config.json @@ -0,0 +1,15 @@ +{ + "server_url": "https://keywarden.dev.ntbx.io/api/v1", + "server_id": "6", + "agent_api_token": "HV8MnbysmINShu9wkXzzNrQE5NyRlPJHfP2aU6SX4HERrlac8EQpoMEu02FX1TYv", + "sync_interval_seconds": 5, + "log_batch_size": 2000, + "state_dir": "/var/lib/keywarden-agent", + "account_policy": { + "username_template": "{{username}}_{{user_id}}", + "default_shell": "/bin/bash", + "admin_group": "sudo", + "create_home": true, + "lock_on_revoke": true + } +} diff --git a/agent/internal/client/client.go b/agent/internal/client/client.go index 0320869..69daf42 100644 --- a/agent/internal/client/client.go +++ b/agent/internal/client/client.go @@ -28,6 +28,18 @@ type Client struct { scheme string host string addr string + token string +} + +func normalizeAgentToken(raw string) string { + token := strings.TrimSpace(raw) + if token == "" { + return "" + } + if len(token) > 7 && strings.EqualFold(token[:7], "Bearer ") { + token = strings.TrimSpace(token[7:]) + } + return token } func New(cfg *config.Config) (*Client, error) { @@ -35,6 +47,10 @@ func New(cfg *config.Config) (*Client, error) { if baseURL == "" { return nil, errors.New("server url is required") } + agentToken := normalizeAgentToken(cfg.AgentAPIToken) + if agentToken == "" { + return nil, errors.New("agent api token is required (set agent_api_token or KEYWARDEN_AGENT_API_TOKEN)") + } cert, err := tls.LoadX509KeyPair(cfg.ClientCertPath(), cfg.ClientKeyPath()) if err != nil { return nil, fmt.Errorf("load client cert: %w", err) @@ -97,9 +113,17 @@ func New(cfg *config.Config) (*Client, error) { scheme: scheme, host: host, addr: addr, + token: agentToken, }, nil } +func (c *Client) setAuth(req *http.Request) { + if c.token == "" { + return + } + req.Header.Set("Authorization", "Bearer "+c.token) +} + type EnrollRequest struct { Token string `json:"token"` CSRPEM string `json:"csr_pem"` @@ -113,6 +137,7 @@ type EnrollResponse struct { ServerID string `json:"server_id"` ClientCert string `json:"client_cert_pem"` CACert string `json:"ca_cert_pem"` + AgentToken string `json:"agent_api_token,omitempty"` SyncProfile string `json:"sync_profile,omitempty"` DisplayName string `json:"display_name,omitempty"` } @@ -149,6 +174,19 @@ type SyncReportRequest struct { Accounts []AccountSyncEntry `json:"accounts,omitempty"` } +type LogSourceConfig struct { + SourceID string `json:"source_id"` + Kind string `json:"kind"` + Name string `json:"name"` + ServiceUnit string `json:"service_unit,omitempty"` + FilePath string `json:"file_path,omitempty"` + Parser string `json:"parser,omitempty"` + IncludeMatches map[string][]string `json:"include_matches,omitempty"` + ExcludeMatches map[string][]string `json:"exclude_matches,omitempty"` + Category string `json:"category"` + EventType string `json:"event_type"` +} + func Enroll(ctx context.Context, serverURL string, req EnrollRequest) (*EnrollResponse, error) { baseURL := strings.TrimRight(serverURL, "/") if baseURL == "" { @@ -241,6 +279,7 @@ func (c *Client) FetchAccountAccess(ctx context.Context, serverID string) ([]Acc if err != nil { return nil, fmt.Errorf("build account access request: %w", err) } + c.setAuth(req) resp, err := c.http.Do(req) if err != nil { return nil, fmt.Errorf("fetch account access: %w", err) @@ -266,6 +305,7 @@ func (c *Client) FetchUserCA(ctx context.Context, serverID string) (*UserCARespo if err != nil { return nil, fmt.Errorf("build user ca request: %w", err) } + c.setAuth(req) resp, err := c.http.Do(req) if err != nil { return nil, fmt.Errorf("fetch user ca: %w", err) @@ -299,6 +339,7 @@ func (c *Client) SendSyncReport(ctx context.Context, serverID string, report Syn return fmt.Errorf("build sync report: %w", err) } req.Header.Set("Content-Type", "application/json") + c.setAuth(req) resp, err := c.http.Do(req) if err != nil { return fmt.Errorf("send sync report: %w", err) @@ -310,12 +351,39 @@ func (c *Client) SendSyncReport(ctx context.Context, serverID string, report Syn return nil } +func (c *Client) FetchLogConfig(ctx context.Context, serverID string) ([]LogSourceConfig, error) { + req, err := http.NewRequestWithContext( + ctx, + http.MethodGet, + c.baseURL+"/agent/servers/"+serverID+"/log-config", + nil, + ) + if err != nil { + return nil, fmt.Errorf("build log config request: %w", err) + } + c.setAuth(req) + resp, err := c.http.Do(req) + if err != nil { + return nil, fmt.Errorf("fetch log config: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode >= 300 { + return nil, &HTTPStatusError{StatusCode: resp.StatusCode, Status: resp.Status} + } + var out []LogSourceConfig + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return nil, fmt.Errorf("decode log config: %w", err) + } + return out, nil +} + func (c *Client) SendLogBatch(ctx context.Context, serverID string, payload []byte) error { req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/agent/servers/"+serverID+"/logs", bytes.NewReader(payload)) if err != nil { return fmt.Errorf("build log request: %w", err) } req.Header.Set("Content-Type", "application/json") + c.setAuth(req) resp, err := c.http.Do(req) if err != nil { return fmt.Errorf("send log batch: %w", err) @@ -344,6 +412,7 @@ func (c *Client) UpdateHost(ctx context.Context, serverID string, reqBody Heartb return fmt.Errorf("build host update: %w", err) } req.Header.Set("Content-Type", "application/json") + c.setAuth(req) resp, err := c.http.Do(req) if err != nil { return fmt.Errorf("send host update: %w", err) diff --git a/agent/internal/config/config.go b/agent/internal/config/config.go index 1701ab1..2995005 100644 --- a/agent/internal/config/config.go +++ b/agent/internal/config/config.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "os" + "strconv" "strings" ) @@ -26,12 +27,28 @@ type AccountPolicy struct { LockOnRevoke bool `json:"lock_on_revoke"` } +type LogSource struct { + SourceID string `json:"source_id,omitempty"` + Kind string `json:"kind"` + Name string `json:"name,omitempty"` + ServiceUnit string `json:"service_unit,omitempty"` + FilePath string `json:"file_path,omitempty"` + Parser string `json:"parser,omitempty"` + IncludeMatches map[string][]string `json:"include_matches,omitempty"` + ExcludeMatches map[string][]string `json:"exclude_matches,omitempty"` + Category string `json:"category,omitempty"` + EventType string `json:"event_type,omitempty"` + Enabled *bool `json:"enabled,omitempty"` +} + type Config struct { ServerURL string `json:"server_url"` ServerID string `json:"server_id,omitempty"` + AgentAPIToken string `json:"agent_api_token,omitempty"` ServerCAPath string `json:"server_ca_path,omitempty"` SyncIntervalSeconds int `json:"sync_interval_seconds,omitempty"` LogBatchSize int `json:"log_batch_size,omitempty"` + LogSources []LogSource `json:"log_sources,omitempty"` StateDir string `json:"state_dir,omitempty"` AccountPolicy AccountPolicy `json:"account_policy,omitempty"` } @@ -48,7 +65,11 @@ func LoadOrInit(path string, serverURL string) (*Config, error) { if serverURL == "" { return nil, errors.New("server url required for first boot") } - cfg := &Config{ServerURL: serverURL, ServerCAPath: os.Getenv("KEYWARDEN_SERVER_CA_PATH")} + cfg := &Config{ + ServerURL: serverURL, + AgentAPIToken: os.Getenv("KEYWARDEN_AGENT_API_TOKEN"), + ServerCAPath: os.Getenv("KEYWARDEN_SERVER_CA_PATH"), + } applyDefaults(cfg) if err := validate(cfg, false); err != nil { return nil, err @@ -62,9 +83,15 @@ func LoadOrInit(path string, serverURL string) (*Config, error) { if err := json.Unmarshal(data, cfg); err != nil { return nil, fmt.Errorf("parse config: %w", err) } + envAgentToken := strings.TrimSpace(os.Getenv("KEYWARDEN_AGENT_API_TOKEN")) if cfg.ServerCAPath == "" { cfg.ServerCAPath = os.Getenv("KEYWARDEN_SERVER_CA_PATH") } + // Environment should win when set so operators can rotate tokens centrally + // without editing persisted agent.json on every node. + if envAgentToken != "" { + cfg.AgentAPIToken = envAgentToken + } applyDefaults(cfg) if err := validate(cfg, false); err != nil { return nil, err @@ -108,6 +135,8 @@ func applyDefaults(cfg *Config) { } func validate(cfg *Config, requireServerID bool) error { + cfg.ServerID = strings.TrimSpace(cfg.ServerID) + cfg.AgentAPIToken = strings.TrimSpace(cfg.AgentAPIToken) var missing []string if cfg.ServerURL == "" { missing = append(missing, "server_url") @@ -118,12 +147,57 @@ func validate(cfg *Config, requireServerID bool) error { if len(missing) > 0 { return fmt.Errorf("missing required config fields: %v", missing) } + if looksLikePlaceholder(cfg.ServerID) { + return errors.New("invalid server_id placeholder in config; remove server_id and re-enroll the agent") + } + if cfg.ServerID != "" { + if _, err := strconv.Atoi(cfg.ServerID); err != nil { + return errors.New("invalid server_id in config; expected numeric server id from enrollment") + } + } + if looksLikePlaceholder(cfg.AgentAPIToken) { + return errors.New("invalid agent_api_token placeholder in config; remove agent_api_token and re-enroll the agent") + } if cfg.SyncIntervalSeconds < 5 { return errors.New("sync_interval_seconds must be >= 5") } + for i, source := range cfg.LogSources { + kind := strings.ToLower(strings.TrimSpace(source.Kind)) + switch kind { + case "journal": + // journal source is valid with optional include/exclude matches. + case "service": + if strings.TrimSpace(source.ServiceUnit) == "" && strings.TrimSpace(source.Name) == "" { + return fmt.Errorf("log_sources[%d]: service source requires service_unit or name", i) + } + case "file": + if strings.TrimSpace(source.FilePath) == "" && strings.TrimSpace(source.Name) == "" { + return fmt.Errorf("log_sources[%d]: file source requires file_path or name", i) + } + default: + return fmt.Errorf("log_sources[%d]: kind must be 'journal', 'service' or 'file'", i) + } + parser := strings.ToLower(strings.TrimSpace(source.Parser)) + if parser == "" { + continue + } + switch parser { + case "none", "syslog", "nginx_access", "json": + default: + return fmt.Errorf("log_sources[%d]: parser must be one of none|syslog|nginx_access|json", i) + } + } return nil } +func looksLikePlaceholder(value string) bool { + value = strings.TrimSpace(value) + if value == "" { + return false + } + return strings.Contains(value, "<") || strings.Contains(value, ">") +} + func (c *Config) ClientCertPath() string { return c.StateDir + "/agent.crt" } @@ -144,6 +218,28 @@ func (c *Config) LogSpoolDir() string { return c.StateDir + "/spool" } +func (c *Config) LogOffsetsPath() string { + return c.StateDir + "/log.offsets.json" +} + +func (c *Config) LogStatePath() string { + return c.StateDir + "/log.state.json" +} + +func (c *Config) EnabledLogSources() []LogSource { + if len(c.LogSources) == 0 { + return nil + } + out := make([]LogSource, 0, len(c.LogSources)) + for _, source := range c.LogSources { + if source.Enabled != nil && !*source.Enabled { + continue + } + out = append(out, source) + } + return out +} + func dir(path string) string { if idx := strings.LastIndex(path, string(os.PathSeparator)); idx != -1 { return path[:idx] diff --git a/agent/internal/logs/collector.go b/agent/internal/logs/collector.go index 1d4e534..c8606f5 100644 --- a/agent/internal/logs/collector.go +++ b/agent/internal/logs/collector.go @@ -1,89 +1,977 @@ package logs import ( + "bufio" "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "math" + "os" + "regexp" + "sort" + "strconv" "strings" + "syscall" "time" "github.com/coreos/go-systemd/v22/sdjournal" ) const defaultLimit = 500 +const initialFileTailBytes = 1 << 20 -type Collector struct { - matches []string +type Collector struct{} + +type journalSource struct { + ID string + Kind string + Unit string + Name string + Parser string + Category string + EventType string + IncludeMatches map[string][]string + ExcludeMatches map[string][]string } +type fileSource struct { + ID string + Path string + Name string + Parser string + Category string + EventType string +} + +var ( + syslogLinePattern = regexp.MustCompile(`^([A-Z][a-z]{2}\s+\d+\s+\d\d:\d\d:\d\d)\s+(\S+)\s+([^\[:]+)(?:\[(\d+)\])?:\s*(.*)$`) + nginxAccessPattern = regexp.MustCompile(`^(\S+)\s+\S+\s+\S+\s+\[([^\]]+)\]\s+"([A-Z]+)\s+([^\s"]+)\s+HTTP/[^"]+"\s+(\d{3})\s+(\S+)`) + numericTimePattern = regexp.MustCompile(`^-?\d+(?:\.\d+)?$`) +) + func NewCollector() *Collector { - return &Collector{matches: defaultMatches()} + return &Collector{} } -func (c *Collector) Collect(ctx context.Context, cursor string, limit int) ([]Event, string, error) { +func (c *Collector) Collect( + ctx context.Context, + limit int, + sources []SourceConfig, + state map[string]SourceState, +) ([]Event, map[string]SourceState, error) { if limit <= 0 { limit = defaultLimit } + journalSources, fileSources := normalizeSources(sources) + nextState := cloneSourceState(state) + events := make([]Event, 0, limit) + seen := map[string]struct{}{} + + collectJournalByDefault := len(journalSources) == 0 && len(fileSources) == 0 + if collectJournalByDefault { + journalSources = []journalSource{{ + ID: "default-journal", + Kind: "journal", + Name: "journal", + Parser: "none", + Category: "system", + EventType: "system", + }} + } + + for _, source := range journalSources { + if len(events) >= limit { + break + } + select { + case <-ctx.Done(): + return events, nextState, ctx.Err() + default: + } + current := nextState[source.ID] + sourceEvents, sourceState, err := c.collectJournalSource(ctx, source, current, limit-len(events), collectJournalByDefault) + if err != nil { + return nil, state, err + } + nextState[source.ID] = sourceState + events = appendUniqueEvents(events, sourceEvents, seen, limit) + } + + if collectJournalByDefault && len(events) == 0 { + source := journalSources[0] + current := nextState[source.ID] + if current.Cursor == "" { + sourceEvents, sourceState, err := c.collectJournalSource(ctx, source, current, limit-len(events), false) + if err == nil { + nextState[source.ID] = sourceState + events = appendUniqueEvents(events, sourceEvents, seen, limit) + } + } + } + + if len(events) >= limit || len(fileSources) == 0 { + return events, nextState, nil + } + + var firstFileErr error + for _, source := range fileSources { + if len(events) >= limit { + break + } + select { + case <-ctx.Done(): + return events, nextState, ctx.Err() + default: + } + current := nextState[source.ID] + fileEvents, sourceState, err := collectFileSource(source, current, limit-len(events)) + if err != nil { + if firstFileErr == nil { + firstFileErr = fmt.Errorf("collect file source %s: %w", source.Path, err) + } + continue + } + nextState[source.ID] = sourceState + events = appendUniqueEvents(events, fileEvents, seen, limit) + } + if len(events) == 0 && firstFileErr != nil { + return events, nextState, firstFileErr + } + + return events, nextState, nil +} + +func (c *Collector) collectJournalSource( + ctx context.Context, + source journalSource, + state SourceState, + limit int, + currentBootOnly bool, +) ([]Event, SourceState, error) { + if limit <= 0 { + return []Event{}, state, nil + } j, err := sdjournal.NewJournal() if err != nil { - return nil, "", err + return nil, state, err } defer j.Close() - for i, match := range c.matches { - if i > 0 { - if err := j.AddDisjunction(); err != nil { - return nil, "", err - } + bootID := currentBootID() + if currentBootOnly && bootID != "" { + if err := j.AddMatch("_BOOT_ID=" + bootID); err != nil { + return nil, state, err } - if err := j.AddMatch(match); err != nil { - return nil, "", err + } + if source.Unit != "" { + if err := j.AddMatch("_SYSTEMD_UNIT=" + source.Unit); err != nil { + return nil, state, err } } - if cursor != "" { - if err := j.SeekCursor(cursor); err == nil { - _, _ = j.Next() - } - } else { - _ = j.SeekTail() - _, _ = j.Next() - } + // Always collect from tail and stop at the last seen cursor to prioritize + // newest logs and avoid head-of-journal backlog blocking recent events. + return c.collectJournalFromTail(ctx, j, source, state, limit, currentBootOnly, bootID) +} - var events []Event - var nextCursor string +func (c *Collector) collectJournalFromTail( + ctx context.Context, + j *sdjournal.Journal, + source journalSource, + state SourceState, + limit int, + currentBootOnly bool, + bootID string, +) ([]Event, SourceState, error) { + if err := j.SeekTail(); err != nil { + return nil, state, err + } + events := make([]Event, 0, limit) + nextState := state + newestCursor := "" + stopCursor := strings.TrimSpace(state.Cursor) for len(events) < limit { select { case <-ctx.Done(): - return events, nextCursor, ctx.Err() + return events, nextState, ctx.Err() default: } - n, err := j.Next() + n, err := j.Previous() if err != nil { - return events, nextCursor, err + return events, nextState, err } if n == 0 { break } entry, err := j.GetEntry() if err != nil { - return events, nextCursor, err + return events, nextState, err } - event := fromEntry(entry) - events = append(events, event) - nextCursor = entry.Cursor + if newestCursor == "" { + newestCursor = entry.Cursor + } + if stopCursor != "" && entry.Cursor == stopCursor { + break + } + if !journalEntryMatchesSource(entry.Fields, source, currentBootOnly, bootID) { + continue + } + events = append(events, buildJournalEvent(entry, source)) } - - return events, nextCursor, nil + if newestCursor != "" { + nextState.Cursor = newestCursor + } + return events, nextState, nil } -func defaultMatches() []string { - return []string{ - "_SYSTEMD_UNIT=sshd.service", - "_SYSTEMD_UNIT=sudo.service", - "_SYSTEMD_UNIT=systemd-networkd.service", - "_SYSTEMD_UNIT=NetworkManager.service", - "_SYSTEMD_UNIT=systemd-logind.service", - "_TRANSPORT=kernel", +func buildJournalEvent(entry *sdjournal.JournalEntry, source journalSource) Event { + event := fromEntry(entry) + event.SourceKind = source.Kind + event.SourceName = source.Name + if source.Category != "" { + event.Category = source.Category } + if event.EventType == "" { + if source.EventType != "" { + event.EventType = source.EventType + } else { + event.EventType = defaultEventType(event.Category) + } + } + if applied := applyParser(&event, source.Parser); applied { + if event.EventType == "" { + event.EventType = source.EventType + } + } + return event +} + +func journalEntryMatchesSource(fields map[string]string, source journalSource, currentBootOnly bool, bootID string) bool { + if currentBootOnly && bootID != "" { + if !strings.EqualFold(strings.TrimSpace(fields["_BOOT_ID"]), strings.TrimSpace(bootID)) { + return false + } + } + if source.Unit != "" && !strings.EqualFold(strings.TrimSpace(fields["_SYSTEMD_UNIT"]), strings.TrimSpace(source.Unit)) { + return false + } + for key, values := range source.IncludeMatches { + if !fieldMatchesAny(fields[key], values) { + return false + } + } + for key, values := range source.ExcludeMatches { + if fieldMatchesAny(fields[key], values) { + return false + } + } + return true +} + +func fieldMatchesAny(fieldValue string, values []string) bool { + fieldValue = strings.TrimSpace(fieldValue) + if fieldValue == "" { + return false + } + for _, candidate := range values { + if strings.EqualFold(fieldValue, strings.TrimSpace(candidate)) { + return true + } + } + return false +} + +func collectFileSource(source fileSource, state SourceState, limit int) ([]Event, SourceState, error) { + if limit <= 0 { + return []Event{}, state, nil + } + file, err := os.Open(source.Path) + if err != nil { + if os.IsNotExist(err) { + next := state + next.Offset = 0 + next.Inode = 0 + return []Event{}, next, nil + } + return nil, state, err + } + defer file.Close() + + info, err := file.Stat() + if err != nil { + return nil, state, err + } + size := info.Size() + inode := fileInode(info) + offset := state.Offset + if offset < 0 { + offset = 0 + } + if state.Inode == 0 && state.Offset == 0 && size > initialFileTailBytes { + offset = size - initialFileTailBytes + } + if state.Inode != 0 && inode != 0 && state.Inode != inode { + offset = 0 + } + if offset > size { + offset = 0 + } + if _, err := file.Seek(offset, io.SeekStart); err != nil { + return nil, state, err + } + + reader := bufio.NewReader(file) + current := offset + if offset > 0 { + skipped, skipErr := reader.ReadString('\n') + current += int64(len(skipped)) + if skipErr != nil && skipErr != io.EOF { + next := state + next.Offset = current + next.Inode = inode + return nil, next, skipErr + } + } + events := make([]Event, 0, limit) + + for len(events) < limit { + line, err := reader.ReadString('\n') + if len(line) > 0 { + lineStart := current + current += int64(len(line)) + message := strings.TrimRight(line, "\r\n") + if message != "" { + event := NewEvent(time.Now()) + event.SourceKind = "file" + event.SourceName = source.Name + event.Category = source.Category + event.EventType = source.EventType + event.Message = compactWhitespace(message) + event.Raw = message + event.Fields = map[string]string{ + "file_path": source.Path, + "file_inode": strconv.FormatUint(inode, 10), + "file_offset": strconv.FormatInt(lineStart, 10), + } + _ = applyParser(&event, source.Parser) + events = append(events, event) + } + } + if err == io.EOF { + break + } + if err != nil { + next := state + next.Offset = current + next.Inode = inode + return events, next, err + } + } + + next := state + next.Offset = current + next.Inode = inode + reverseEvents(events) + return events, next, nil +} + +func normalizeSources(sources []SourceConfig) ([]journalSource, []fileSource) { + journalSources := make([]journalSource, 0, len(sources)) + fileSources := make([]fileSource, 0, len(sources)) + for _, source := range sources { + kind := strings.ToLower(strings.TrimSpace(source.Kind)) + id := sourceStateID(source) + parser := normalizeParser(source.Parser) + includeMatches := normalizeMatches(source.IncludeMatches) + excludeMatches := normalizeMatches(source.ExcludeMatches) + + switch kind { + case "service": + unit := strings.TrimSpace(source.ServiceUnit) + if unit == "" { + unit = strings.TrimSpace(source.Name) + } + if unit == "" { + continue + } + category := strings.TrimSpace(source.Category) + if category == "" { + category = defaultCategoryForService(unit) + } + eventType := strings.TrimSpace(source.EventType) + if eventType == "" { + eventType = defaultEventType(category) + } + name := strings.TrimSpace(source.Name) + if name == "" { + name = unit + } + includeMatches["_SYSTEMD_UNIT"] = appendUniqueStrings(includeMatches["_SYSTEMD_UNIT"], unit) + journalSources = append(journalSources, journalSource{ + ID: id, + Kind: "service", + Unit: unit, + Name: name, + Parser: parser, + Category: category, + EventType: eventType, + IncludeMatches: includeMatches, + ExcludeMatches: excludeMatches, + }) + case "journal": + category := strings.TrimSpace(source.Category) + if category == "" { + category = "system" + } + eventType := strings.TrimSpace(source.EventType) + if eventType == "" { + eventType = defaultEventType(category) + } + name := strings.TrimSpace(source.Name) + if name == "" { + name = "journal" + } + journalSources = append(journalSources, journalSource{ + ID: id, + Kind: "journal", + Unit: strings.TrimSpace(source.ServiceUnit), + Name: name, + Parser: parser, + Category: category, + EventType: eventType, + IncludeMatches: includeMatches, + ExcludeMatches: excludeMatches, + }) + case "file": + path := strings.TrimSpace(source.FilePath) + if path == "" { + path = strings.TrimSpace(source.Name) + } + if path == "" { + continue + } + category := strings.TrimSpace(source.Category) + if category == "" { + category = "file" + } + eventType := strings.TrimSpace(source.EventType) + if eventType == "" { + eventType = "file.line" + } + name := strings.TrimSpace(source.Name) + if name == "" { + name = path + } + fileSources = append(fileSources, fileSource{ + ID: id, + Path: path, + Name: name, + Parser: parser, + Category: category, + EventType: eventType, + }) + } + } + return journalSources, fileSources +} + +func sourceStateID(source SourceConfig) string { + explicit := strings.TrimSpace(source.SourceID) + if explicit != "" { + return explicit + } + parts := []string{ + strings.ToLower(strings.TrimSpace(source.Kind)), + strings.TrimSpace(source.Name), + strings.TrimSpace(source.ServiceUnit), + strings.TrimSpace(source.FilePath), + strings.ToLower(strings.TrimSpace(source.Parser)), + strings.TrimSpace(source.Category), + strings.TrimSpace(source.EventType), + encodeMatchMap(source.IncludeMatches), + encodeMatchMap(source.ExcludeMatches), + } + sum := sha256.Sum256([]byte(strings.Join(parts, "\x1f"))) + return "src-" + hex.EncodeToString(sum[:8]) +} + +func normalizeParser(parser string) string { + switch strings.ToLower(strings.TrimSpace(parser)) { + case "syslog", "nginx_access", "json": + return strings.ToLower(strings.TrimSpace(parser)) + default: + return "none" + } +} + +func normalizeMatches(in map[string][]string) map[string][]string { + if len(in) == 0 { + return map[string][]string{} + } + out := map[string][]string{} + for key, values := range in { + keyText := strings.TrimSpace(key) + if keyText == "" { + continue + } + cleanValues := make([]string, 0, len(values)) + for _, value := range values { + valueText := strings.TrimSpace(value) + if valueText == "" { + continue + } + cleanValues = append(cleanValues, valueText) + } + if len(cleanValues) == 0 { + continue + } + out[keyText] = appendUniqueStrings(nil, cleanValues...) + } + return out +} + +func appendUniqueStrings(base []string, values ...string) []string { + seen := map[string]struct{}{} + out := append([]string{}, base...) + for _, value := range out { + seen[strings.ToLower(strings.TrimSpace(value))] = struct{}{} + } + for _, value := range values { + normalized := strings.TrimSpace(value) + if normalized == "" { + continue + } + key := strings.ToLower(normalized) + if _, exists := seen[key]; exists { + continue + } + seen[key] = struct{}{} + out = append(out, normalized) + } + sort.Strings(out) + return out +} + +func cloneSourceState(in map[string]SourceState) map[string]SourceState { + out := map[string]SourceState{} + for key, value := range in { + out[key] = value + } + return out +} + +func appendUniqueEvents(dst []Event, src []Event, seen map[string]struct{}, limit int) []Event { + for _, event := range src { + if len(dst) >= limit { + break + } + key := eventDedupKey(event) + if key != "" { + if _, exists := seen[key]; exists { + continue + } + seen[key] = struct{}{} + } + dst = append(dst, event) + } + return dst +} + +func reverseEvents(events []Event) { + for left, right := 0, len(events)-1; left < right; left, right = left+1, right-1 { + events[left], events[right] = events[right], events[left] + } +} + +func eventDedupKey(event Event) string { + if len(event.Fields) > 0 { + if cursor := strings.TrimSpace(event.Fields["journal_cursor"]); cursor != "" { + return "journal:" + cursor + } + path := strings.TrimSpace(event.Fields["file_path"]) + inode := strings.TrimSpace(event.Fields["file_inode"]) + offset := strings.TrimSpace(event.Fields["file_offset"]) + if path != "" && offset != "" { + return "file:" + path + ":" + inode + ":" + offset + } + } + if event.SourceKind != "" && event.SourceName != "" && event.Timestamp != "" && event.Message != "" { + return strings.Join([]string{"fallback", event.SourceKind, event.SourceName, event.Timestamp, event.Message}, "|") + } + return "" +} + +func applyParser(event *Event, parser string) bool { + switch normalizeParser(parser) { + case "json": + return applyJSONParser(event) + case "nginx_access": + return applyNginxAccessParser(event) + case "syslog": + return applySyslogParser(event) + default: + return false + } +} + +func applyJSONParser(event *Event) bool { + payload := map[string]any{} + if err := json.Unmarshal([]byte(event.Message), &payload); err != nil { + return false + } + if event.Fields == nil { + event.Fields = map[string]string{} + } + for key, value := range payload { + keyText := strings.TrimSpace(key) + if keyText == "" { + continue + } + event.Fields["json."+keyText] = fmt.Sprint(value) + } + if parsedTimestamp, ok := parseJSONTimestamp(payload); ok { + maybeSetParsedTimestamp(event, parsedTimestamp) + } + for _, candidate := range []string{"message", "msg", "log"} { + if value, ok := payload[candidate]; ok { + message := compactWhitespace(fmt.Sprint(value)) + if message != "" { + event.Message = message + break + } + } + } + if event.Username == "" { + for _, candidate := range []string{"username", "user"} { + if value, ok := payload[candidate]; ok { + event.Username = strings.TrimSpace(fmt.Sprint(value)) + if event.Username != "" { + break + } + } + } + } + if event.SourceIP == "" { + for _, candidate := range []string{"source_ip", "ip", "remote_addr", "client_ip"} { + if value, ok := payload[candidate]; ok { + event.SourceIP = strings.TrimSpace(fmt.Sprint(value)) + if event.SourceIP != "" { + break + } + } + } + } + return true +} + +func applyNginxAccessParser(event *Event) bool { + matches := nginxAccessPattern.FindStringSubmatch(event.Message) + if len(matches) == 0 { + return false + } + if parsedTimestamp, err := time.Parse("02/Jan/2006:15:04:05 -0700", strings.TrimSpace(matches[2])); err == nil { + maybeSetParsedTimestamp(event, parsedTimestamp) + } + if event.SourceIP == "" { + event.SourceIP = strings.TrimSpace(matches[1]) + } + if event.Fields == nil { + event.Fields = map[string]string{} + } + event.Fields["nginx.time_local"] = matches[2] + event.Fields["nginx.method"] = matches[3] + event.Fields["nginx.path"] = matches[4] + event.Fields["nginx.status"] = matches[5] + event.Fields["nginx.body_bytes_sent"] = matches[6] + if event.EventType == "" || event.EventType == "file.line" || event.EventType == "system" { + event.EventType = "http.access" + } + if event.Category == "system" { + event.Category = "access" + } + return true +} + +func applySyslogParser(event *Event) bool { + message := strings.TrimSpace(event.Message) + if message == "" { + return false + } + if strings.HasPrefix(message, "<") { + if end := strings.Index(message, ">"); end > 1 { + priorityText := strings.TrimSpace(message[1:end]) + if priorityValue, err := strconv.Atoi(priorityText); err == nil { + severity := priorityValue % 8 + if event.Priority == "" { + event.Priority = strconv.Itoa(severity) + } + message = strings.TrimSpace(message[end+1:]) + } + } + } + matches := syslogLinePattern.FindStringSubmatch(message) + if len(matches) == 0 { + return false + } + if parsedTimestamp, ok := parseSyslogTimestamp(matches[1], time.Now()); ok { + maybeSetParsedTimestamp(event, parsedTimestamp) + } + if event.Hostname == "" { + event.Hostname = strings.TrimSpace(matches[2]) + } + identifier := strings.TrimSpace(matches[3]) + pid := strings.TrimSpace(matches[4]) + parsedMessage := compactWhitespace(matches[5]) + if parsedMessage != "" { + event.Message = parsedMessage + } + if event.Fields == nil { + event.Fields = map[string]string{} + } + if identifier != "" { + event.Fields["syslog_identifier"] = identifier + } + if pid != "" { + event.Fields["syslog_pid"] = pid + } + if event.SourceName == "" && identifier != "" { + event.SourceName = identifier + } + if event.SourceIP == "" || event.Username == "" { + _, username, sourceIP, _ := parseMessage(event.Category, event.Message) + if event.Username == "" { + event.Username = username + } + if event.SourceIP == "" { + event.SourceIP = sourceIP + } + } + return true +} + +func maybeSetParsedTimestamp(event *Event, timestamp time.Time) { + if timestamp.IsZero() { + return + } + // Journal entries already provide canonical timestamps from journald. + if event.SourceKind != "" && event.SourceKind != "file" { + return + } + event.Timestamp = timestamp.UTC().Format(time.RFC3339Nano) +} + +func parseJSONTimestamp(payload map[string]any) (time.Time, bool) { + for _, candidate := range []string{ + "@timestamp", + "timestamp", + "time", + "ts", + "datetime", + "date", + "event_time", + } { + value, ok := payload[candidate] + if !ok { + continue + } + if parsed, ok := parseTimestampValue(value); ok { + return parsed, true + } + } + return time.Time{}, false +} + +func parseTimestampValue(value any) (time.Time, bool) { + switch typed := value.(type) { + case string: + return parseTimestampString(typed) + case float64: + return parseNumericTimestamp(strconv.FormatFloat(typed, 'f', -1, 64)) + case json.Number: + return parseNumericTimestamp(typed.String()) + case int: + return parseNumericTimestamp(strconv.FormatInt(int64(typed), 10)) + case int8: + return parseNumericTimestamp(strconv.FormatInt(int64(typed), 10)) + case int16: + return parseNumericTimestamp(strconv.FormatInt(int64(typed), 10)) + case int32: + return parseNumericTimestamp(strconv.FormatInt(int64(typed), 10)) + case int64: + return parseNumericTimestamp(strconv.FormatInt(typed, 10)) + case uint: + return parseNumericTimestamp(strconv.FormatUint(uint64(typed), 10)) + case uint8: + return parseNumericTimestamp(strconv.FormatUint(uint64(typed), 10)) + case uint16: + return parseNumericTimestamp(strconv.FormatUint(uint64(typed), 10)) + case uint32: + return parseNumericTimestamp(strconv.FormatUint(uint64(typed), 10)) + case uint64: + if typed > math.MaxInt64 { + return time.Time{}, false + } + return parseNumericTimestamp(strconv.FormatUint(typed, 10)) + default: + return time.Time{}, false + } +} + +func parseTimestampString(value string) (time.Time, bool) { + candidate := strings.TrimSpace(value) + if candidate == "" { + return time.Time{}, false + } + if numericTimePattern.MatchString(candidate) { + if parsed, ok := parseNumericTimestamp(candidate); ok { + return parsed, true + } + } + + layoutsWithZone := []string{ + time.RFC3339Nano, + time.RFC3339, + "2006-01-02 15:04:05.999999999Z07:00", + "2006-01-02 15:04:05Z07:00", + "02/Jan/2006:15:04:05 -0700", + time.RFC1123Z, + time.RFC1123, + time.RFC822Z, + time.RFC822, + time.RFC850, + time.RubyDate, + time.UnixDate, + } + for _, layout := range layoutsWithZone { + if parsed, err := time.Parse(layout, candidate); err == nil { + return parsed, true + } + } + + layoutsWithoutZone := []string{ + "2006-01-02 15:04:05.999999999", + "2006-01-02 15:04:05,999", + "2006-01-02 15:04:05", + "2006-01-02T15:04:05.999999999", + "2006-01-02T15:04:05", + "2006-01-02", + time.ANSIC, + } + for _, layout := range layoutsWithoutZone { + if parsed, err := time.ParseInLocation(layout, candidate, time.Local); err == nil { + return parsed, true + } + } + return time.Time{}, false +} + +func parseNumericTimestamp(value string) (time.Time, bool) { + candidate := strings.TrimSpace(value) + if candidate == "" { + return time.Time{}, false + } + if strings.Contains(candidate, ".") { + numericValue, err := strconv.ParseFloat(candidate, 64) + if err != nil { + return time.Time{}, false + } + return parseUnixFloatTimestamp(numericValue), true + } + numericValue, err := strconv.ParseInt(candidate, 10, 64) + if err != nil { + return time.Time{}, false + } + return parseUnixIntTimestamp(numericValue), true +} + +func parseUnixIntTimestamp(value int64) time.Time { + absValue := value + if absValue < 0 { + absValue = -absValue + } + switch { + case absValue >= 1e18: + return time.Unix(0, value) + case absValue >= 1e15: + return time.Unix(0, value*int64(time.Microsecond)) + case absValue >= 1e12: + return time.Unix(0, value*int64(time.Millisecond)) + default: + return time.Unix(value, 0) + } +} + +func parseUnixFloatTimestamp(value float64) time.Time { + absValue := math.Abs(value) + switch { + case absValue >= 1e18: + return time.Unix(0, int64(value)) + case absValue >= 1e15: + return time.Unix(0, int64(value*float64(time.Microsecond))) + case absValue >= 1e12: + return time.Unix(0, int64(value*float64(time.Millisecond))) + default: + seconds, fraction := math.Modf(value) + return time.Unix(int64(seconds), int64(fraction*float64(time.Second))) + } +} + +func parseSyslogTimestamp(value string, now time.Time) (time.Time, bool) { + candidate := compactWhitespace(value) + if candidate == "" { + return time.Time{}, false + } + if now.IsZero() { + now = time.Now() + } + nowLocal := now.In(time.Local) + parsed, err := time.ParseInLocation("Jan 2 15:04:05 2006", fmt.Sprintf("%s %d", candidate, nowLocal.Year()), time.Local) + if err != nil { + return time.Time{}, false + } + // Syslog timestamps omit the year; pull back one year if parsing lands + // in the future due to year rollover. + if parsed.After(nowLocal.Add(24 * time.Hour)) { + parsed = parsed.AddDate(-1, 0, 0) + } + return parsed, true +} + +func fileInode(info os.FileInfo) uint64 { + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok { + return 0 + } + return stat.Ino +} + +func encodeMatchMap(in map[string][]string) string { + if len(in) == 0 { + return "" + } + keys := make([]string, 0, len(in)) + for key := range in { + keys = append(keys, key) + } + sort.Strings(keys) + parts := make([]string, 0, len(keys)) + for _, key := range keys { + values := append([]string{}, in[key]...) + sort.Strings(values) + parts = append(parts, key+"="+strings.Join(values, ",")) + } + return strings.Join(parts, ";") +} + +func currentBootID() string { + data, err := os.ReadFile("/proc/sys/kernel/random/boot_id") + if err != nil { + return "" + } + return strings.TrimSpace(string(data)) } func fromEntry(entry *sdjournal.JournalEntry) Event { @@ -91,26 +979,89 @@ func fromEntry(entry *sdjournal.JournalEntry) Event { event := NewEvent(ts) fields := entry.Fields unit := fields["_SYSTEMD_UNIT"] - message := fields["MESSAGE"] + rawMessage := strings.TrimSpace(fields["MESSAGE"]) identifier := fields["SYSLOG_IDENTIFIER"] + pid := fields["_PID"] event.Unit = unit - event.Message = message + event.Message = compactJournalLine(identifier, pid, rawMessage) + if rawMessage != "" && rawMessage != event.Message { + event.Raw = rawMessage + } event.Priority = fields["PRIORITY"] event.Hostname = fields["_HOSTNAME"] - event.Fields = fields + event.Fields = compactJournalFields(fields, entry.Cursor) event.Category = categorize(unit, identifier, fields) - event.EventType, event.Username, event.SourceIP, event.SessionID = parseMessage(event.Category, message) + event.EventType, event.Username, event.SourceIP, event.SessionID = parseMessage(event.Category, rawMessage) if event.EventType == "" { event.EventType = defaultEventType(event.Category) } return event } +func compactJournalLine(identifier string, pid string, message string) string { + normalizedMessage := compactWhitespace(message) + identifier = strings.TrimSpace(identifier) + pid = strings.TrimSpace(pid) + + if normalizedMessage == "" { + if identifier != "" && pid != "" { + return identifier + "[" + pid + "]" + } + return identifier + } + if identifier != "" && pid != "" { + return identifier + "[" + pid + "]: " + normalizedMessage + } + if identifier != "" { + return identifier + ": " + normalizedMessage + } + return normalizedMessage +} + +func compactWhitespace(value string) string { + if value == "" { + return "" + } + return strings.Join(strings.Fields(value), " ") +} + +func compactJournalFields(fields map[string]string, cursor string) map[string]string { + compact := map[string]string{} + if cursor != "" { + compact["journal_cursor"] = cursor + } + if bootID := strings.TrimSpace(fields["_BOOT_ID"]); bootID != "" { + compact["boot_id"] = bootID + } + if transport := strings.TrimSpace(fields["_TRANSPORT"]); transport != "" { + compact["transport"] = transport + } + if comm := strings.TrimSpace(fields["_COMM"]); comm != "" { + compact["comm"] = comm + } + if exe := strings.TrimSpace(fields["_EXE"]); exe != "" { + compact["exe"] = exe + } + if pid := strings.TrimSpace(fields["_PID"]); pid != "" { + compact["pid"] = pid + } + if uid := strings.TrimSpace(fields["_UID"]); uid != "" { + compact["uid"] = uid + } + if gid := strings.TrimSpace(fields["_GID"]); gid != "" { + compact["gid"] = gid + } + if len(compact) == 0 { + return nil + } + return compact +} + func categorize(unit string, identifier string, fields map[string]string) string { switch { - case unit == "sshd.service" || identifier == "sshd": + case unit == "sshd.service" || unit == "ssh.service" || identifier == "sshd": return "access" case unit == "sudo.service" || identifier == "sudo": return "auth" @@ -123,6 +1074,19 @@ func categorize(unit string, identifier string, fields map[string]string) string } } +func defaultCategoryForService(unit string) string { + switch strings.ToLower(unit) { + case "sshd.service", "ssh.service": + return "access" + case "sudo.service": + return "auth" + case "systemd-networkd.service", "networkmanager.service": + return "network" + default: + return "system" + } +} + func defaultEventType(category string) string { switch category { case "access": @@ -131,6 +1095,8 @@ func defaultEventType(category string) string { return "auth" case "network": return "network" + case "file": + return "file.line" default: return "system" } diff --git a/agent/internal/logs/collector_parser_test.go b/agent/internal/logs/collector_parser_test.go new file mode 100644 index 0000000..59a82c6 --- /dev/null +++ b/agent/internal/logs/collector_parser_test.go @@ -0,0 +1,60 @@ +package logs + +import ( + "testing" + "time" +) + +func TestApplyNginxAccessParserSetsParsedTimestamp(t *testing.T) { + event := NewEvent(time.Date(2026, time.April, 15, 18, 43, 7, 0, time.UTC)) + event.SourceKind = "file" + event.Message = `192.168.2.40 - - [28/Jan/2026:17:43:43 +0000] "GET / HTTP/1.1" 200 612 "-" "curl/8.0"` + + if !applyNginxAccessParser(&event) { + t.Fatalf("expected nginx parser to match log line") + } + + expected := "2026-01-28T17:43:43Z" + if event.Timestamp != expected { + t.Fatalf("expected parsed timestamp %s, got %s", expected, event.Timestamp) + } +} + +func TestParseJSONTimestampSupportsEpochMillis(t *testing.T) { + event := NewEvent(time.Date(2026, time.April, 15, 18, 43, 7, 0, time.UTC)) + event.SourceKind = "file" + event.Message = `{"timestamp":1700000000000,"message":"login event"}` + + if !applyJSONParser(&event) { + t.Fatalf("expected json parser to parse payload") + } + + expected := time.Unix(1700000000, 0).UTC().Format(time.RFC3339Nano) + if event.Timestamp != expected { + t.Fatalf("expected parsed timestamp %s, got %s", expected, event.Timestamp) + } +} + +func TestParseSyslogTimestampHandlesYearRollover(t *testing.T) { + now := time.Date(2026, time.January, 1, 0, 0, 30, 0, time.UTC) + parsed, ok := parseSyslogTimestamp("Dec 31 23:59:59", now) + if !ok { + t.Fatalf("expected syslog timestamp to parse") + } + if parsed.Year() != 2025 { + t.Fatalf("expected year rollover to previous year, got %d", parsed.Year()) + } +} + +func TestMaybeSetParsedTimestampSkipsJournalAndServiceSources(t *testing.T) { + original := "2026-04-15T18:43:07Z" + event := Event{ + Timestamp: original, + SourceKind: "service", + } + + maybeSetParsedTimestamp(&event, time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)) + if event.Timestamp != original { + t.Fatalf("expected service timestamp to remain unchanged") + } +} diff --git a/agent/internal/logs/offsets.go b/agent/internal/logs/offsets.go new file mode 100644 index 0000000..1506001 --- /dev/null +++ b/agent/internal/logs/offsets.go @@ -0,0 +1,35 @@ +package logs + +import ( + "encoding/json" + "os" +) + +func ReadOffsets(path string) (map[string]int64, error) { + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return map[string]int64{}, nil + } + return nil, err + } + out := map[string]int64{} + if len(data) == 0 { + return out, nil + } + if err := json.Unmarshal(data, &out); err != nil { + return nil, err + } + return out, nil +} + +func WriteOffsets(path string, offsets map[string]int64) error { + if offsets == nil { + offsets = map[string]int64{} + } + data, err := json.Marshal(offsets) + if err != nil { + return err + } + return os.WriteFile(path, data, 0o600) +} diff --git a/agent/internal/logs/state.go b/agent/internal/logs/state.go new file mode 100644 index 0000000..d63cf42 --- /dev/null +++ b/agent/internal/logs/state.go @@ -0,0 +1,35 @@ +package logs + +import ( + "encoding/json" + "os" +) + +func ReadState(path string) (map[string]SourceState, error) { + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return map[string]SourceState{}, nil + } + return nil, err + } + out := map[string]SourceState{} + if len(data) == 0 { + return out, nil + } + if err := json.Unmarshal(data, &out); err != nil { + return nil, err + } + return out, nil +} + +func WriteState(path string, state map[string]SourceState) error { + if state == nil { + state = map[string]SourceState{} + } + data, err := json.Marshal(state) + if err != nil { + return err + } + return os.WriteFile(path, data, 0o600) +} diff --git a/agent/internal/logs/types.go b/agent/internal/logs/types.go index db6e961..15eefab 100644 --- a/agent/internal/logs/types.go +++ b/agent/internal/logs/types.go @@ -3,21 +3,42 @@ package logs import "time" type Event struct { - Timestamp string `json:"timestamp"` - Category string `json:"category"` - EventType string `json:"event_type"` - Unit string `json:"unit,omitempty"` - Priority string `json:"priority,omitempty"` - Hostname string `json:"hostname,omitempty"` - Username string `json:"username,omitempty"` - Principal string `json:"principal,omitempty"` - SourceIP string `json:"source_ip,omitempty"` - SessionID string `json:"session_id,omitempty"` - Message string `json:"message,omitempty"` - Raw string `json:"raw,omitempty"` - Fields map[string]string `json:"fields,omitempty"` + Timestamp string `json:"timestamp"` + Category string `json:"category"` + EventType string `json:"event_type"` + SourceKind string `json:"source_kind,omitempty"` + SourceName string `json:"source_name,omitempty"` + Unit string `json:"unit,omitempty"` + Priority string `json:"priority,omitempty"` + Hostname string `json:"hostname,omitempty"` + Username string `json:"username,omitempty"` + Principal string `json:"principal,omitempty"` + SourceIP string `json:"source_ip,omitempty"` + SessionID string `json:"session_id,omitempty"` + Message string `json:"message,omitempty"` + Raw string `json:"raw,omitempty"` + Fields map[string]string `json:"fields,omitempty"` } func NewEvent(ts time.Time) Event { return Event{Timestamp: ts.UTC().Format(time.RFC3339Nano)} } + +type SourceConfig struct { + SourceID string + Kind string + Name string + ServiceUnit string + FilePath string + Parser string + IncludeMatches map[string][]string + ExcludeMatches map[string][]string + Category string + EventType string +} + +type SourceState struct { + Cursor string `json:"cursor,omitempty"` + Offset int64 `json:"offset,omitempty"` + Inode uint64 `json:"inode,omitempty"` +} diff --git a/agent/keywarden-agent b/agent/keywarden-agent index 63711ec..51f726c 100755 Binary files a/agent/keywarden-agent and b/agent/keywarden-agent differ diff --git a/app/apps/access/permissions.py b/app/apps/access/permissions.py index 1a7a6ef..5df8aa4 100644 --- a/app/apps/access/permissions.py +++ b/app/apps/access/permissions.py @@ -17,6 +17,7 @@ def sync_server_view_perm(access_request: AccessRequest) -> None: server_id=access_request.server_id, status=AccessRequest.Status.APPROVED, ) + .filter(Q(request_shell=True) | Q(request_logs=True) | Q(request_users=True)) .filter(Q(expires_at__isnull=True) | Q(expires_at__gt=now)) .exists() ) diff --git a/app/apps/accounts/templates/accounts/profile.html b/app/apps/accounts/templates/accounts/profile.html index 92b8df9..9a6a06f 100644 --- a/app/apps/accounts/templates/accounts/profile.html +++ b/app/apps/accounts/templates/accounts/profile.html @@ -205,6 +205,32 @@