diff --git a/log-input/config/config.go b/log-input/config/config.go index f7277dec0..22942dbbd 100644 --- a/log-input/config/config.go +++ b/log-input/config/config.go @@ -14,6 +14,11 @@ type Config struct { // behind the auth interceptors, so a probe there would need credentials. HealthAddr string + // HTTPListenAddr is the TLS HTTP ingest endpoint (/v1/ingest). + // Kept separate from the gRPC port so each transport can be load-balanced + // or firewalled independently. + HTTPListenAddr string + CertFile string KeyFile string @@ -38,8 +43,9 @@ type Config struct { } const ( - defaultListenAddr = "0.0.0.0:50051" - defaultHealthAddr = "0.0.0.0:8080" + defaultListenAddr = "0.0.0.0:50051" + defaultHealthAddr = "0.0.0.0:8080" + defaultHTTPListenAddr = "0.0.0.0:50052" defaultShards = 16 defaultTenant = "ce66672c-e36d-4761-a8c8-90058fee1a24" defaultAuthTTL = 5 * time.Minute @@ -54,8 +60,9 @@ func Load() (*Config, error) { certs := envOr("CERTS_FOLDER", defaultCertsFolder) c := &Config{ - ListenAddr: envOr("LISTEN_ADDR", defaultListenAddr), - HealthAddr: envOr("HEALTH_ADDR", defaultHealthAddr), + ListenAddr: envOr("LISTEN_ADDR", defaultListenAddr), + HealthAddr: envOr("HEALTH_ADDR", defaultHealthAddr), + HTTPListenAddr: envOr("HTTP_LISTEN_ADDR", defaultHTTPListenAddr), CertFile: certs + "/" + utmCertFileName, KeyFile: certs + "/" + utmCertFileKeyName, NATSURL: os.Getenv("NATS_URL"), @@ -90,6 +97,9 @@ func Load() (*Config, error) { if c.AuthTTL < minAuthTTL { return nil, fmt.Errorf("AUTH_TTL must be at least %s", minAuthTTL) } + if c.HTTPListenAddr == "" { + return nil, fmt.Errorf("HTTP_LISTEN_ADDR must not be empty") + } return c, nil } diff --git a/log-input/ingest/http.go b/log-input/ingest/http.go new file mode 100644 index 000000000..bc047659e --- /dev/null +++ b/log-input/ingest/http.go @@ -0,0 +1,238 @@ +package ingest + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net" + "net/http" + "strconv" + "time" + + "github.com/nats-io/nats.go" + "github.com/threatwinds/go-sdk/catcher" + "github.com/threatwinds/go-sdk/plugins" + + "github.com/utmstack/UTMStack/log-input/config" +) + +const ( + maxBodyBytes = 10 << 20 // 10 MB + maxBatchSize = 1000 +) + +type ingestResponse struct { + Accepted int `json:"accepted"` + Failed []failedEntry `json:"failed"` +} + +type failedEntry struct { + Index int `json:"index"` + Reason string `json:"reason"` +} + +func allErrorsAreNATSDown(errs []error) bool { + if len(errs) == 0 { + return false + } + for _, err := range errs { + if !isNATSDown(err) { + return false + } + } + return true +} + +func isNATSDown(err error) bool { + return errors.Is(err, nats.ErrConnectionClosed) || + errors.Is(err, nats.ErrNoResponders) +} + +type HTTPServer struct { + srv *http.Server + resolver tenantResolver + pub Publisher + cfg *config.Config +} + +func NewHTTPServer(cfg *config.Config, resolver tenantResolver, pub Publisher) (*HTTPServer, error) { + tlsCfg, err := loadTLSConfig(cfg.CertFile, cfg.KeyFile) + if err != nil { + return nil, catcher.Error("cannot load TLS config for HTTP server", err, map[string]any{ + "process": processName, + "cert": cfg.CertFile, + }) + } + + h := &HTTPServer{ + resolver: resolver, + pub: pub, + cfg: cfg, + } + + mux := http.NewServeMux() + mux.HandleFunc("/v1/ingest", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + h.handle(w, r) + }) + + h.srv = &http.Server{ + Addr: cfg.HTTPListenAddr, + Handler: mux, + TLSConfig: tlsCfg, + ReadTimeout: 30 * time.Second, + WriteTimeout: 35 * time.Second, + } + + return h, nil +} + +func (h *HTTPServer) Serve() error { + catcher.Info("http ingest listening", map[string]any{ + "process": processName, + "addr": h.cfg.HTTPListenAddr, + }) + return h.srv.ListenAndServeTLS("", "") +} + +func (h *HTTPServer) Stop(ctx context.Context) error { + return h.srv.Shutdown(ctx) +} + +func (h *HTTPServer) handle(w http.ResponseWriter, r *http.Request) { + defer func() { + if rec := recover(); rec != nil { + _ = catcher.Error("panic in http ingest handler", fmt.Errorf("%v", rec), map[string]any{ + "process": processName, + }) + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "internal server error"}) + } + }() + + logs, err := parseBody(w, r) + if err != nil { + return + } + + clientIP, _, _ := net.SplitHostPort(r.RemoteAddr) + creds := Credentials{ + APIKey: r.Header.Get("Utm-Api-Key"), + ConnKey: r.Header.Get("X-Connector-Key"), + ConnType: r.Header.Get("X-Connector-Type"), + ClientIP: clientIP, + } + if connIDStr := r.Header.Get("X-Connector-Id"); connIDStr != "" { + creds.ConnID, err = strconv.ParseUint(connIDStr, 10, 64) + if err != nil { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "X-Connector-Id is not a valid uint64"}) + return + } + } + + tenant, err := resolveAuth(r.Context(), h.resolver, creds) + if err != nil { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": err.Error()}) + return + } + + ctx := WithTenant(r.Context(), tenant) + + var ( + accepted int + failed []failedEntry + pubErrs []error + ) + + for i, l := range logs { + applyDefaults(ctx, h.cfg, l) + + if pubErr := h.pub.Publish(ctx, l); pubErr != nil { + failed = append(failed, failedEntry{Index: i, Reason: pubErr.Error()}) + pubErrs = append(pubErrs, pubErr) + continue + } + accepted++ + } + + if accepted == 0 && len(failed) > 0 && allErrorsAreNATSDown(pubErrs) { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "broker unavailable"}) + return + } + + if failed == nil { + failed = []failedEntry{} + } + writeJSON(w, http.StatusOK, ingestResponse{Accepted: accepted, Failed: failed}) +} + +func parseBody(w http.ResponseWriter, r *http.Request) ([]*plugins.Log, error) { + r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes) + + var buf []byte + { + dec := json.NewDecoder(r.Body) + var raw json.RawMessage + if err := dec.Decode(&raw); err != nil { + var maxErr *http.MaxBytesError + if errors.As(err, &maxErr) { + writeJSON(w, http.StatusRequestEntityTooLarge, map[string]string{"error": "body exceeds 10 MB limit"}) + return nil, err + } + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()}) + return nil, err + } + buf = []byte(raw) + } + + var shapeProbe struct { + Logs *[]json.RawMessage `json:"logs"` + } + if err := json.Unmarshal(buf, &shapeProbe); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()}) + return nil, err + } + + if shapeProbe.Logs != nil { + rawLogs := *shapeProbe.Logs + if len(rawLogs) == 0 { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "empty batch"}) + return nil, fmt.Errorf("empty batch") + } + if len(rawLogs) > maxBatchSize { + writeJSON(w, http.StatusBadRequest, map[string]string{ + "error": fmt.Sprintf("batch exceeds %d logs", maxBatchSize), + }) + return nil, fmt.Errorf("batch exceeds %d logs", maxBatchSize) + } + logs := make([]*plugins.Log, 0, len(rawLogs)) + for i, rawEntry := range rawLogs { + var l plugins.Log + if err := json.Unmarshal(rawEntry, &l); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{ + "error": fmt.Sprintf("log[%d]: invalid JSON: %s", i, err.Error()), + }) + return nil, err + } + logs = append(logs, &l) + } + return logs, nil + } + + var l plugins.Log + if err := json.Unmarshal(buf, &l); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()}) + return nil, err + } + return []*plugins.Log{&l}, nil +} + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(v) +} diff --git a/log-input/ingest/middleware.go b/log-input/ingest/middleware.go index c15a5e80d..d21926a11 100644 --- a/log-input/ingest/middleware.go +++ b/log-input/ingest/middleware.go @@ -2,6 +2,7 @@ package ingest import ( "context" + "errors" "net" "strconv" @@ -16,14 +17,57 @@ import ( const apiKeyHeader = "Utm-Api-Key" +var errUnauthenticated = errors.New("auth is not provided") + +var errPermissionDenied = errors.New("invalid credential") + +type tenantResolver interface { + APIKeyTenant(ctx context.Context, apiKey, clientIP string) (string, bool) + ConnectorTenant(ctx context.Context, key string, id uint64, typ string) (string, bool) +} + +type grpcResolver interface { + tenantResolver + InternalKeyValid(key string) bool +} + type middlewares struct { - auth *auth.Service + auth grpcResolver } func newMiddlewares(a *auth.Service) *middlewares { return &middlewares{auth: a} } +type Credentials struct { + APIKey string + ConnKey string + ConnID uint64 + ConnType string + ClientIP string +} + +func resolveAuth(ctx context.Context, resolver tenantResolver, c Credentials) (tenant string, err error) { + switch { + case c.APIKey != "": + t, ok := resolver.APIKeyTenant(ctx, c.APIKey, c.ClientIP) + if !ok { + return "", errPermissionDenied + } + return t, nil + + case c.ConnKey != "" && c.ConnID != 0 && c.ConnType != "": + t, ok := resolver.ConnectorTenant(ctx, c.ConnKey, c.ConnID, c.ConnType) + if !ok { + return "", errPermissionDenied + } + return t, nil + + default: + return "", errUnauthenticated + } +} + func (m *middlewares) unary(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) { ctx, err := m.check(ctx) if err != nil { @@ -53,40 +97,41 @@ func (m *middlewares) check(ctx context.Context) (context.Context, error) { return nil, status.Error(codes.Internal, "metadata is not provided") } - authKey := md.Get("key") - authID := md.Get("id") - connectorType := md.Get("type") - apiKey := md.Get(apiKeyHeader) - internalKey := md.Get("internal-key") + if keys := md.Get("internal-key"); len(keys) > 0 { + if !m.auth.InternalKeyValid(keys[0]) { + return nil, status.Error(codes.PermissionDenied, "internal key does not match") + } + return ctx, nil + } - switch { - case len(authKey) > 0 && len(authID) > 0 && len(connectorType) > 0: - id, err := strconv.ParseUint(authID[0], 10, 64) + creds := Credentials{ClientIP: peerIP(ctx)} + if v := md.Get("key"); len(v) > 0 { + creds.ConnKey = v[0] + } + if v := md.Get("id"); len(v) > 0 { + id, err := strconv.ParseUint(v[0], 10, 64) if err != nil { return nil, status.Error(codes.PermissionDenied, "id is not valid") } - tenant, ok := m.auth.ConnectorTenant(ctx, authKey[0], id, connectorType[0]) - if !ok { - return nil, status.Error(codes.PermissionDenied, "invalid key") - } - return WithTenant(ctx, tenant), nil - - case len(apiKey) > 0: - tenant, ok := m.auth.APIKeyTenant(ctx, apiKey[0], peerIP(ctx)) - if !ok { - return nil, status.Error(codes.PermissionDenied, "invalid api key") - } - return WithTenant(ctx, tenant), nil + creds.ConnID = id + } + if v := md.Get("type"); len(v) > 0 { + creds.ConnType = v[0] + } + if v := md.Get(apiKeyHeader); len(v) > 0 { + creds.APIKey = v[0] + } - case len(internalKey) > 0: - if !m.auth.InternalKeyValid(internalKey[0]) { - return nil, status.Error(codes.PermissionDenied, "internal key does not match") + tenant, err := resolveAuth(ctx, m.auth, creds) + if err != nil { + switch { + case errors.Is(err, errUnauthenticated): + return nil, status.Error(codes.Unauthenticated, err.Error()) + default: + return nil, status.Error(codes.PermissionDenied, err.Error()) } - return ctx, nil - - default: - return nil, status.Error(codes.Unauthenticated, "auth is not provided") } + return WithTenant(ctx, tenant), nil } type tenantKey struct{} diff --git a/log-input/ingest/server.go b/log-input/ingest/server.go index 28c75accb..6468debdb 100644 --- a/log-input/ingest/server.go +++ b/log-input/ingest/server.go @@ -16,23 +16,37 @@ import ( "github.com/utmstack/UTMStack/log-input/auth" "github.com/utmstack/UTMStack/log-input/config" - "github.com/utmstack/UTMStack/log-input/publish" ) +func loadTLSConfig(certFile, keyFile string) (*tls.Config, error) { + cert, err := tls.LoadX509KeyPair(certFile, keyFile) + if err != nil { + return nil, err + } + return &tls.Config{ + Certificates: []tls.Certificate{cert}, + MinVersion: tls.VersionTLS13, + }, nil +} + const processName = "log-input" +type Publisher interface { + Publish(ctx context.Context, l *plugins.Log) error +} + type Server struct { plugins.UnimplementedIntegrationServer cfg *config.Config - pub *publish.Publisher + pub Publisher grpc *grpc.Server health *health.Server } -func NewServer(cfg *config.Config, a *auth.Service, pub *publish.Publisher) (*Server, error) { - cert, err := tls.LoadX509KeyPair(cfg.CertFile, cfg.KeyFile) +func NewServer(cfg *config.Config, a *auth.Service, pub Publisher) (*Server, error) { + tlsCfg, err := loadTLSConfig(cfg.CertFile, cfg.KeyFile) if err != nil { return nil, catcher.Error("cannot read the certificate files", err, map[string]any{ "process": processName, @@ -40,10 +54,7 @@ func NewServer(cfg *config.Config, a *auth.Service, pub *publish.Publisher) (*Se }) } - creds := credentials.NewTLS(&tls.Config{ - Certificates: []tls.Certificate{cert}, - MinVersion: tls.VersionTLS13, - }) + creds := credentials.NewTLS(tlsCfg) m := newMiddlewares(a) @@ -100,7 +111,7 @@ func (s *Server) ProcessLog(srv plugins.Integration_ProcessLogServer) error { return err } - s.applyDefaults(ctx, l) + applyDefaults(ctx, s.cfg, l) if err := s.pub.Publish(ctx, l); err != nil { return catcher.Error("cannot publish the log", err, map[string]any{ @@ -119,7 +130,7 @@ func (s *Server) ProcessLog(srv plugins.Integration_ProcessLogServer) error { } } -func (s *Server) applyDefaults(ctx context.Context, l *plugins.Log) { +func applyDefaults(ctx context.Context, cfg *config.Config, l *plugins.Log) { if l.Id == "" { l.Id = uuid.NewString() } @@ -127,7 +138,7 @@ func (s *Server) applyDefaults(ctx context.Context, l *plugins.Log) { l.TenantId = tenant } if l.TenantId == "" { - l.TenantId = s.cfg.DefaultTenant + l.TenantId = cfg.DefaultTenant } if l.DataType == "" { l.DataType = "generic" diff --git a/log-input/ingest/tenant_test.go b/log-input/ingest/tenant_test.go index d081d303a..c297d9d17 100644 --- a/log-input/ingest/tenant_test.go +++ b/log-input/ingest/tenant_test.go @@ -9,18 +9,19 @@ import ( "github.com/utmstack/UTMStack/log-input/config" ) -func newServer() *Server { - return &Server{cfg: &config.Config{DefaultTenant: "default-tenant"}} +// testCfg is a minimal config used by the tenant tests. +// Only DefaultTenant is relevant here; the other fields are deliberately zero. +func testCfg() *config.Config { + return &config.Config{DefaultTenant: "default-tenant"} } // A connector that names someone else's tenant writes into its own. This is the // whole reason the tenant comes from the credential. func TestPushedTenantCannotBeChosen(t *testing.T) { - s := newServer() ctx := WithTenant(context.Background(), "tenant-a") l := &plugins.Log{TenantId: "tenant-b"} - s.applyDefaults(ctx, l) + applyDefaults(ctx, testCfg(), l) if l.TenantId != "tenant-a" { t.Errorf("tenant = %q, want the credential's tenant", l.TenantId) @@ -28,11 +29,10 @@ func TestPushedTenantCannotBeChosen(t *testing.T) { } func TestTenantIsTakenFromTheCredential(t *testing.T) { - s := newServer() ctx := WithTenant(context.Background(), "tenant-a") l := &plugins.Log{} - s.applyDefaults(ctx, l) + applyDefaults(ctx, testCfg(), l) if l.TenantId != "tenant-a" { t.Errorf("tenant = %q, want tenant-a", l.TenantId) @@ -42,10 +42,8 @@ func TestTenantIsTakenFromTheCredential(t *testing.T) { // The internal key belongs to UTMStack's own services, which push on behalf of // every tenant, so what they send stands. func TestInternalCallerKeepsTheTenantItSent(t *testing.T) { - s := newServer() - l := &plugins.Log{TenantId: "tenant-b"} - s.applyDefaults(context.Background(), l) + applyDefaults(context.Background(), testCfg(), l) if l.TenantId != "tenant-b" { t.Errorf("tenant = %q, want tenant-b", l.TenantId) @@ -53,10 +51,8 @@ func TestInternalCallerKeepsTheTenantItSent(t *testing.T) { } func TestNoTenantAnywhereFallsBackToTheDefault(t *testing.T) { - s := newServer() - l := &plugins.Log{} - s.applyDefaults(context.Background(), l) + applyDefaults(context.Background(), testCfg(), l) if l.TenantId != "default-tenant" { t.Errorf("tenant = %q, want the default", l.TenantId) diff --git a/log-input/main.go b/log-input/main.go index 5d30d82bc..38c9cb967 100644 --- a/log-input/main.go +++ b/log-input/main.go @@ -2,6 +2,8 @@ package main import ( "context" + "errors" + "net/http" "os" "os/signal" "syscall" @@ -57,11 +59,23 @@ func main() { os.Exit(1) } + httpServer, err := ingest.NewHTTPServer(cfg, authService, publisher) + if err != nil { + _ = catcher.Error("cannot build the http ingest server", err, map[string]any{"process": processName}) + time.Sleep(5 * time.Second) + os.Exit(1) + } + healthEndpoint := ingest.NewHealth(cfg.HealthAddr) go healthEndpoint.Serve() errs := make(chan error, 1) go func() { errs <- server.Serve() }() + go func() { + if err := httpServer.Serve(); err != nil && !errors.Is(err, http.ErrServerClosed) { + _ = catcher.Error("the http ingest server stopped", err, map[string]any{"process": processName}) + } + }() signals := make(chan os.Signal, 1) signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM) @@ -79,6 +93,7 @@ func main() { server.Stop() sCtx, sCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + _ = httpServer.Stop(sCtx) healthEndpoint.Close(sCtx) sCancel() }