ITWorx Pulse release export
bd774932d5
Public source validation / validate (push) Failing after 3m8s
204 lines
9.1 KiB
Go
204 lines
9.1 KiB
Go
// Package servicedefaults installs the small, explicitly configured service
|
|
// monitoring baseline used by a production Pulse deployment.
|
|
package servicedefaults
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"path"
|
|
"strings"
|
|
|
|
"github.com/itworx/pulse/internal/probe"
|
|
"github.com/itworx/pulse/internal/reconciliation"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
const identityNamespace = "itworx-pulse-service-defaults"
|
|
|
|
type Options struct {
|
|
PublicURL string
|
|
OIDCIssuer string
|
|
}
|
|
|
|
type Summary struct {
|
|
Services int
|
|
Endpoints int
|
|
Probes int
|
|
Dependencies int
|
|
}
|
|
|
|
type serviceSeed struct {
|
|
id, name, description string
|
|
endpointID string
|
|
endpointName string
|
|
endpointType string
|
|
target probe.Target
|
|
probes []probeSeed
|
|
}
|
|
|
|
type probeSeed struct {
|
|
id, name, kind string
|
|
target probe.Target
|
|
interval int
|
|
timeout int
|
|
expected []int
|
|
verifyTLS bool
|
|
}
|
|
|
|
// Seed validates every configured URL against the same network policy used at
|
|
// execution time, then atomically installs or refreshes only Pulse-owned rows.
|
|
func Seed(ctx context.Context, pool *pgxpool.Pool, options Options, policy probe.NetworkPolicy, resolver probe.Resolver) (Summary, error) {
|
|
if pool == nil {
|
|
return Summary{}, errors.New("service defaults require a database pool")
|
|
}
|
|
seeds, err := desired(ctx, options, policy, resolver)
|
|
if err != nil {
|
|
return Summary{}, err
|
|
}
|
|
if len(seeds) == 0 {
|
|
return Summary{}, nil
|
|
}
|
|
tx, err := pool.BeginTx(ctx, pgx.TxOptions{})
|
|
if err != nil {
|
|
return Summary{}, fmt.Errorf("begin service defaults: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
|
|
summary := Summary{}
|
|
for _, service := range seeds {
|
|
labels := []byte(`{"managedBy":"pulse-system-defaults"}`)
|
|
if _, err := tx.Exec(ctx, `INSERT INTO services (id,name,description,state,labels,revision)
|
|
VALUES ($1,$2,$3,'unknown',$4::jsonb,1)
|
|
ON CONFLICT (id) DO UPDATE SET name=EXCLUDED.name,description=EXCLUDED.description,
|
|
labels=EXCLUDED.labels,archived_at=NULL,updated_at=now(),revision=services.revision+1
|
|
WHERE (services.name,services.description,services.labels,services.archived_at) IS DISTINCT FROM
|
|
(EXCLUDED.name,EXCLUDED.description,EXCLUDED.labels,EXCLUDED.archived_at)`,
|
|
service.id, service.name, service.description, labels); err != nil {
|
|
return Summary{}, fmt.Errorf("upsert system service %q: %w", service.name, err)
|
|
}
|
|
summary.Services++
|
|
target, err := json.Marshal(service.target)
|
|
if err != nil {
|
|
return Summary{}, fmt.Errorf("encode endpoint target: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO service_endpoints (id,service_id,name,endpoint_type,target,enabled,revision)
|
|
VALUES ($1,$2,$3,$4,$5::jsonb,true,1)
|
|
ON CONFLICT (id) DO UPDATE SET name=EXCLUDED.name,endpoint_type=EXCLUDED.endpoint_type,
|
|
target=EXCLUDED.target,enabled=true,archived_at=NULL,updated_at=now(),revision=service_endpoints.revision+1
|
|
WHERE (service_endpoints.name,service_endpoints.endpoint_type,service_endpoints.target,service_endpoints.enabled,service_endpoints.archived_at) IS DISTINCT FROM
|
|
(EXCLUDED.name,EXCLUDED.endpoint_type,EXCLUDED.target,EXCLUDED.enabled,EXCLUDED.archived_at)`,
|
|
service.endpointID, service.id, service.endpointName, service.endpointType, target); err != nil {
|
|
return Summary{}, fmt.Errorf("upsert system endpoint %q: %w", service.endpointName, err)
|
|
}
|
|
summary.Endpoints++
|
|
for _, definition := range service.probes {
|
|
target, err := json.Marshal(definition.target)
|
|
if err != nil {
|
|
return Summary{}, fmt.Errorf("encode probe target: %w", err)
|
|
}
|
|
expected, err := json.Marshal(definition.expected)
|
|
if err != nil {
|
|
return Summary{}, fmt.Errorf("encode expected statuses: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO probes
|
|
(id,service_id,endpoint_id,name,probe_type,target,interval_seconds,timeout_seconds,enabled,expected_status_codes,follow_redirects,verify_tls,revision)
|
|
VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7,$8,true,$9::jsonb,false,$10,1)
|
|
ON CONFLICT (id) DO UPDATE SET endpoint_id=EXCLUDED.endpoint_id,name=EXCLUDED.name,
|
|
probe_type=EXCLUDED.probe_type,target=EXCLUDED.target,interval_seconds=EXCLUDED.interval_seconds,
|
|
timeout_seconds=EXCLUDED.timeout_seconds,enabled=true,expected_status_codes=EXCLUDED.expected_status_codes,
|
|
follow_redirects=false,verify_tls=EXCLUDED.verify_tls,archived_at=NULL,updated_at=now(),revision=probes.revision+1
|
|
WHERE (probes.endpoint_id,probes.name,probes.probe_type,probes.target,probes.interval_seconds,probes.timeout_seconds,
|
|
probes.enabled,probes.expected_status_codes,probes.follow_redirects,probes.verify_tls,probes.archived_at) IS DISTINCT FROM
|
|
(EXCLUDED.endpoint_id,EXCLUDED.name,EXCLUDED.probe_type,EXCLUDED.target,EXCLUDED.interval_seconds,EXCLUDED.timeout_seconds,
|
|
EXCLUDED.enabled,EXCLUDED.expected_status_codes,EXCLUDED.follow_redirects,EXCLUDED.verify_tls,EXCLUDED.archived_at)`,
|
|
definition.id, service.id, service.endpointID, definition.name, definition.kind, target,
|
|
definition.interval, definition.timeout, expected, definition.verifyTLS); err != nil {
|
|
return Summary{}, fmt.Errorf("upsert system probe %q: %w", definition.name, err)
|
|
}
|
|
summary.Probes++
|
|
}
|
|
}
|
|
if len(seeds) == 2 {
|
|
dependencyID := stableID("dependency:pulse-authentik")
|
|
if _, err := tx.Exec(ctx, `INSERT INTO service_dependencies
|
|
(id,service_id,depends_on_service_id,relation_type,confidence,confirmed,first_seen_at,last_seen_at)
|
|
VALUES ($1,$2,$3,'depends_on',1,true,now(),now())
|
|
ON CONFLICT (service_id,depends_on_service_id,relation_type) WHERE source_id IS NULL
|
|
DO UPDATE SET confidence=1,confirmed=true,last_seen_at=now(),archived_at=NULL`,
|
|
dependencyID, seeds[0].id, seeds[1].id); err != nil {
|
|
return Summary{}, fmt.Errorf("upsert Pulse authentication dependency: %w", err)
|
|
}
|
|
summary.Dependencies = 1
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return Summary{}, fmt.Errorf("commit service defaults: %w", err)
|
|
}
|
|
return summary, nil
|
|
}
|
|
|
|
func desired(ctx context.Context, options Options, policy probe.NetworkPolicy, resolver probe.Resolver) ([]serviceSeed, error) {
|
|
publicURL := strings.TrimSpace(options.PublicURL)
|
|
issuerURL := strings.TrimSpace(options.OIDCIssuer)
|
|
if publicURL == "" && issuerURL == "" {
|
|
return nil, nil
|
|
}
|
|
if publicURL == "" || issuerURL == "" {
|
|
return nil, errors.New("PULSE_PUBLIC_URL and PULSE_OIDC_ISSUER must both be configured for system service monitoring")
|
|
}
|
|
public, err := validateHTTPS(ctx, policy, resolver, publicURL)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("validate PULSE_PUBLIC_URL for service monitoring: %w", err)
|
|
}
|
|
issuer, err := validateHTTPS(ctx, policy, resolver, issuerURL)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("validate PULSE_OIDC_ISSUER for service monitoring: %w", err)
|
|
}
|
|
publicPort := urlPort(public)
|
|
issuerPort := urlPort(issuer)
|
|
healthTarget := probe.Target{Scheme: "https", Host: public.Hostname(), Port: publicPort, Path: joinURLPath(public.Path, "healthz")}
|
|
issuerTarget := probe.Target{Scheme: "https", Host: issuer.Hostname(), Port: issuerPort, Path: joinURLPath(issuer.Path, ".well-known/openid-configuration")}
|
|
tlsTarget := probe.Target{Scheme: "tls", Host: public.Hostname(), Port: publicPort}
|
|
return []serviceSeed{
|
|
{id: stableID("service:pulse"), name: "ITWorx Pulse", description: "Publieke Pulse-applicatie en API", endpointID: stableID("endpoint:pulse-public"), endpointName: "Publieke HTTPS-endpoint", endpointType: "http", target: healthTarget,
|
|
probes: []probeSeed{{id: stableID("probe:pulse-health"), name: "Pulse health", kind: probe.TypeHTTP, target: healthTarget, interval: 30, timeout: 10, expected: []int{200}, verifyTLS: true}, {id: stableID("probe:pulse-tls"), name: "Pulse TLS-certificaat", kind: probe.TypeTLS, target: tlsTarget, interval: 21600, timeout: 10, verifyTLS: true}}},
|
|
{id: stableID("service:authentik"), name: "Authentik SSO", description: "OIDC-identiteitsprovider voor Pulse", endpointID: stableID("endpoint:authentik-discovery"), endpointName: "OIDC discovery-endpoint", endpointType: "http", target: issuerTarget,
|
|
probes: []probeSeed{{id: stableID("probe:authentik-discovery"), name: "Authentik OIDC discovery", kind: probe.TypeHTTP, target: issuerTarget, interval: 60, timeout: 10, expected: []int{200}, verifyTLS: true}}},
|
|
}, nil
|
|
}
|
|
|
|
func validateHTTPS(ctx context.Context, policy probe.NetworkPolicy, resolver probe.Resolver, raw string) (*url.URL, error) {
|
|
parsed, err := policy.ValidateURL(ctx, resolver, raw)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if parsed.Scheme != "https" || parsed.RawQuery != "" || parsed.User != nil || parsed.Fragment != "" {
|
|
return nil, errors.New("system service URL must be HTTPS without credentials, query parameters, or fragment")
|
|
}
|
|
return parsed, nil
|
|
}
|
|
|
|
func joinURLPath(base, suffix string) string {
|
|
joined := path.Join("/", base, suffix)
|
|
if !strings.HasPrefix(joined, "/") {
|
|
return "/" + joined
|
|
}
|
|
return joined
|
|
}
|
|
|
|
func urlPort(value *url.URL) int {
|
|
if value.Port() == "" {
|
|
return 443
|
|
}
|
|
var port int
|
|
_, _ = fmt.Sscanf(value.Port(), "%d", &port)
|
|
return port
|
|
}
|
|
|
|
func stableID(key string) string {
|
|
return reconciliation.StableEntityID(identityNamespace, "service-default", key)
|
|
}
|