platform

package
v0.33.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 14, 2026 License: GPL-3.0 Imports: 32 Imported by: 0

Documentation

Overview

Package platform provides components for running sensors in platform mode.

Platform sensors are centrally managed by OpenCTEM and execute jobs on behalf of tenants. Unlike tenant sensors that run within a tenant's infrastructure, platform sensors are deployed and operated by the OpenCTEM platform itself.

REQUIRES THE PLATFORM (SaaS) CONTROL PLANE. The lease, poll, job and registration calls in this package target /api/v1/platform/* routes that the open-source OpenCTEM API does not serve; against an OSS API they fail with 404. Self-hosted sensors should use pkg/client together with core.CommandPoller (protocol v2, /api/v2/sensor/*) instead.

Security note: jobs received through JobPoller are handed to the caller-supplied executor as-is. This package does not validate job targets, so the executor MUST apply its own target validation. The OSS command path (core.DefaultCommandExecutor) does this itself via core.ScanTargetPolicy, and pkg/client honors command expiry, so this package is not the only line of defense for either path.

Key components:

  • LeaseManager: Handles K8s-style lease renewal for health monitoring
  • Bootstrapper: Handles sensor registration using bootstrap tokens
  • JobPoller: Long-polls for jobs using /platform/poll endpoint
  • Client: Extended client with platform-specific endpoints

Usage:

// Bootstrap a new platform sensor
bootstrapper := platform.NewBootstrapper(baseURL, bootstrapToken)
creds, err := bootstrapper.Register(ctx, &platform.RegistrationRequest{
    Name: "scanner-001",
    Capabilities: []string{"sast", "sca"},
})

// Create platform client
client := platform.NewClient(&platform.ClientConfig{
    BaseURL: baseURL,
    APIKey:  creds.APIKey,
    SensorID: creds.SensorID,
})

// Start lease manager
leaseManager := platform.NewLeaseManager(client, &platform.LeaseConfig{
    LeaseDuration: 60 * time.Second,
    RenewInterval: 20 * time.Second,
})
go leaseManager.Start(ctx)

// Start job poller
poller := platform.NewJobPoller(client, executor, &platform.PollerConfig{
    MaxJobs:     5,
    PollTimeout: 30 * time.Second,
})
poller.Start(ctx)

Stability: Internal-bound (docs/STABILITY.md): public today because other public packages use it; it moves under internal/ before v1.0.0. Do not import it from a sensor.

Index

Constants

View Source
const (
	DefaultLeaseDuration     = 60 * time.Second
	DefaultRenewInterval     = 20 * time.Second
	DefaultPollTimeout       = 30 * time.Second
	DefaultMaxConcurrentJobs = 5
	DefaultBootstrapTimeout  = 30 * time.Second
)

Default configuration values.

View Source
const (
	// DefaultStateDir is where a sensor keeps its state when nothing else is
	// configured and the directory is usable. The images create it for the
	// sensor user; the install snippets mount one data volume on its parent,
	// /var/lib/openctem (state/, outbox/, content/), and a mount on the
	// state directory itself still works. CheckStatePersistence judges a
	// directory by its nearest mount, so either counts.
	DefaultStateDir = "/var/lib/openctem/state"
	// EnvStateDir overrides the state directory.
	EnvStateDir = "SENSOR_STATE_DIR"
)

State directory defaults.

View Source
const Version = "1.0.0"

Version is the platform package version.

Variables

View Source
var (
	ErrBootstrapTokenInvalid = errors.New("invalid or expired bootstrap token")
	ErrSensorAlreadyExists   = errors.New("sensor with this name already exists")
)

Non-retryable registration errors. A bad bootstrap token or a name conflict will never succeed on retry, so the loop stops immediately on these.

View Source
var ErrCredentialsFileNotMoved = errors.New("credentials file not moved")

ErrCredentialsFileNotMoved is returned by MigrateCredentialsFile when the old file is valid but could not be moved (a read-only volume, a Kubernetes Secret, a single bind-mounted file). Nothing was changed.

View Source
var ErrRenewUnsupported = errors.New("the platform does not serve sensor protocol v2 key renewal " +
	"(upgrade the OpenCTEM API: protocol v1 is retired)")

ErrRenewUnsupported: the platform does not serve sensor protocol v2 key renewal (an OpenCTEM API from before 2026-10); upgrade the platform.

Functions

func DecideKeyAutoRenew added in v0.12.0

func DecideKeyAutoRenew(setting string, p StatePersistence) (bool, string)

DecideKeyAutoRenew turns a renewal setting into a decision: "true" / "on" / "1" renews, "false" / "off" / "0" does not, and "" or "auto" renews only when the state directory persists (a renewed key that is lost on restart locks the sensor out).

func DefaultCredentialsFile added in v0.7.0

func DefaultCredentialsFile() (string, error)

DefaultCredentialsFile is where a sensor keeps its credentials when no path is configured: ~/.openctem/sensor-credentials.json.

func KeyFingerprint added in v0.12.0

func KeyFingerprint(key string) string

KeyFingerprint returns a salted PBKDF2-SHA256 fingerprint of an API key, "pbkdf2-sha256$<iterations>$<salt>$<hash>" (base64url, no padding). The credentials file keeps the fingerprint of the configured key its renewed key descends from, never the configured key itself; a fresh salt makes each fingerprint different, so compare with KeyFingerprintMatches.

func KeyFingerprintMatches added in v0.12.0

func KeyFingerprintMatches(fp, key string) (match, ok bool)

KeyFingerprintMatches reports whether key has the fingerprint fp (KeyFingerprint). ok is false when fp is not a fingerprint this SDK can check (empty, malformed, another scheme), so the caller can treat it as unknown rather than as a different key.

func LegacyCredentialsFile added in v0.7.0

func LegacyCredentialsFile() (string, error)

LegacyCredentialsFile is the default credentials path of sensors from before the agent -> sensor rename: ~/.openctem/agent-credentials.json.

func MigrateCredentialsFile added in v0.7.0

func MigrateCredentialsFile(from, to string) (moved bool, err error)

MigrateCredentialsFile moves the credentials file at from to to, without ever leaving the sensor with no readable credentials:

  1. from is read and parsed (an unreadable or invalid file is an error, not a reason to register a second sensor);
  2. the credentials are written to to atomically — temp file with mode 0600, fsync, rename, directory fsync;
  3. to is read back and compared with what was read from from;
  4. only then is from removed.

An interruption at any step leaves from in place, and the next start migrates again. If writing to, verifying it or removing from fails, to is removed again and the error wraps ErrCredentialsFileNotMoved: the state is exactly as before. When to already exists it wins: from is left untouched and a warning names both files. When from does not exist there is nothing to do. It reports whether a file was moved.

func ResolveCredentialsFile added in v0.7.0

func ResolveCredentialsFile(explicit string) (string, error)

ResolveCredentialsFile returns the credentials file a sensor should use. An explicit path is returned as is. With an empty path it returns DefaultCredentialsFile, after moving a pre-rename credentials file there (MigrateCredentialsFile), so an upgraded sensor keeps its identity and API key without registering again. When the old file cannot be moved (see ErrCredentialsFileNotMoved) it is used where it is, with a warning, so a deployment that mounts it read-only keeps starting exactly as before.

func ResolveStateCredentialsFile added in v0.12.0

func ResolveStateCredentialsFile(explicit, stateDir string) (string, error)

ResolveStateCredentialsFile returns the credentials file a sensor should use. An explicit path is returned as is. Otherwise it is the file in stateDir; a credentials file an earlier version kept in the home directory (~/.openctem/sensor-credentials.json, or the pre-rename agent-credentials.json) is moved there first (MigrateCredentialsFile), so an upgraded sensor keeps the key it renewed. When that file cannot be moved it is used where it is, with a warning.

func ResolveStateDir added in v0.12.0

func ResolveStateDir(explicit string) string

ResolveStateDir returns the sensor's state directory: explicit, else $SENSOR_STATE_DIR, else DefaultStateDir when it exists or can be created and is writable, else ~/.openctem (a sensor binary run by a user without /var/lib/openctem). The returned directory may not exist yet only in the last case.

func RotatedKeySaver added in v0.12.0

func RotatedKeySaver(store *FileCredentialStore, configuredKey, sensorID string) func(newKey string, expiresAt *time.Time) error

RotatedKeySaver returns a KeyRenewConfig.OnRotated that saves each renewed key, its expiry (or that it has none) and the fingerprint of configuredKey (the key the sensor was installed with) to the credentials file, atomically and 0600 (FileCredentialStore.Save).

func StateCredentialsFile added in v0.12.0

func StateCredentialsFile(stateDir string) string

StateCredentialsFile is the credentials file in a state directory.

func ValidateJob

func ValidateJob(job *JobInfo, config *PollerConfig) error

ValidateJob validates a job before execution. SECURITY: This prevents processing of malformed or unauthorized jobs.

Types

type AuditLogger

type AuditLogger interface {
	// JobStarted logs a job start event.
	JobStarted(jobID, jobType string, details map[string]interface{})

	// JobCompleted logs a job completion event.
	JobCompleted(jobID string, duration time.Duration, details map[string]interface{})

	// JobFailed logs a job failure event.
	JobFailed(jobID string, err error, details map[string]interface{})

	// ResourceThrottle logs a resource throttle event.
	ResourceThrottle(reason string, metrics map[string]interface{})
}

AuditLogger is an optional interface for audit logging. If set on JobPoller, job lifecycle events will be logged.

type BootstrapConfig

type BootstrapConfig struct {
	// Timeout for the registration request.
	Timeout time.Duration

	// RetryAttempts is the number of times to retry on failure.
	RetryAttempts int

	// RetryDelay is the delay between retry attempts.
	RetryDelay time.Duration

	// Verbose enables debug logging.
	Verbose bool
}

BootstrapConfig configures the Bootstrapper.

type Bootstrapper

type Bootstrapper struct {
	// contains filtered or unexported fields
}

Bootstrapper handles platform sensor registration using bootstrap tokens.

Bootstrap tokens are short-lived tokens that allow new sensors to register themselves with the platform. The flow is:

  1. Admin creates a bootstrap token via CLI or API
  2. Token is provided to the sensor deployment (e.g., via environment variable)
  3. Sensor uses Bootstrapper to register and receive permanent API credentials
  4. Sensor stores credentials securely and uses them for all future API calls

Example:

bootstrapper := platform.NewBootstrapper(baseURL, os.Getenv("BOOTSTRAP_TOKEN"))
creds, err := bootstrapper.Register(ctx, &platform.RegistrationRequest{
    Name: "scanner-001",
    Capabilities: []string{"sast", "sca"},
    Region: "us-east-1",
})

No server for this flow: no OpenCTEM API serves POST /api/v1/platform/register (bootstrap tokens were never built; api RFC-032 §3.4). Sensors authenticate with the API key an administrator issues; enrollment tokens (api RFC-032 Phase 2) replace this flow. Kept so code that compiles against it keeps compiling, until it is removed.

func NewBootstrapper

func NewBootstrapper(baseURL, bootstrapToken string, config *BootstrapConfig) *Bootstrapper

NewBootstrapper creates a new Bootstrapper.

func (*Bootstrapper) Register

Register registers a new platform sensor and returns the API credentials.

IMPORTANT: The returned API key is only provided once. Store it securely (e.g., in a secrets manager or encrypted file). If lost, the sensor must be deleted and re-registered with a new bootstrap token.

type ClientConfig

type ClientConfig struct {
	// BaseURL is the API base URL.
	BaseURL string

	// APIKey is the sensor's API key.
	APIKey string

	// SensorID is the sensor's ID.
	SensorID string

	// PollTimeout is the long-poll timeout for job polling.
	PollTimeout time.Duration

	// Verbose enables debug logging.
	Verbose bool
}

ClientConfig configures the PlatformClient.

type CredentialStore

type CredentialStore interface {
	Save(creds *SensorCredentials) error
	Load() (*SensorCredentials, error)
	Exists() bool
}

CredentialStore interface for storing sensor credentials.

type EnsureRegisteredConfig

type EnsureRegisteredConfig struct {
	// BaseURL is the API base URL.
	BaseURL string

	// BootstrapToken is the bootstrap token for registration.
	// Only needed if credentials don't exist.
	BootstrapToken string

	// CredentialsFile is the path to store/load credentials. Empty means
	// DefaultCredentialsFile (with the pre-rename file migrated); on return
	// EnsureRegistered has set it to the path actually used.
	CredentialsFile string

	// Registration is the registration request.
	// Only needed if credentials don't exist.
	Registration *RegistrationRequest

	// Verbose enables debug logging.
	Verbose bool
}

EnsureRegisteredConfig configures EnsureRegistered.

type FileCredentialStore

type FileCredentialStore struct {
	Path string
}

FileCredentialStore stores credentials in a file.

func NewFileCredentialStore

func NewFileCredentialStore(path string) *FileCredentialStore

NewFileCredentialStore creates a new file-based credential store.

func (*FileCredentialStore) Exists

func (s *FileCredentialStore) Exists() bool

Exists checks if the credentials file exists.

func (*FileCredentialStore) Load

Load loads credentials from the file.

func (*FileCredentialStore) Save

Save saves credentials to the file.

The write is atomic and owner-only: the JSON goes to a 0600 temp file in the same directory, is fsynced, then renamed over Path. A crash or full disk mid-write therefore never leaves a truncated credentials file (which would lose the only copy of a rotated key), and a pre-existing file with looser permissions is replaced by a 0600 one rather than rewritten in place with its old mode.

type JobClient

type JobClient interface {
	// Poll polls for new jobs using long-polling.
	Poll(ctx context.Context, req *PollRequest) (*PollResponse, error)

	// AcknowledgeJob acknowledges receipt of a job.
	AcknowledgeJob(ctx context.Context, jobID string) error

	// ReportJobResult reports the result of a completed job.
	ReportJobResult(ctx context.Context, result *JobResult) error

	// ReportJobProgress reports progress on a running job.
	ReportJobProgress(ctx context.Context, jobID string, progress int, message string) error
}

JobClient defines the interface for job operations.

func NewHTTPJobClient

func NewHTTPJobClient(baseURL, apiKey, sensorID string, pollTimeout time.Duration) JobClient

NewHTTPJobClient creates a new HTTP-based job client.

type JobExecutor

type JobExecutor interface {
	Execute(ctx context.Context, job *JobInfo) (*JobResult, error)
}

JobExecutor executes platform jobs.

type JobInfo

type JobInfo struct {
	ID         string                 `json:"id"`
	Type       string                 `json:"type"`
	Priority   int                    `json:"priority"`
	TenantID   string                 `json:"tenant_id"`
	Payload    map[string]interface{} `json:"payload"`
	AuthToken  string                 `json:"auth_token"` // JWT for tenant data access
	CreatedAt  time.Time              `json:"created_at"`
	TimeoutSec int                    `json:"timeout_seconds"`

	// WorkflowContext is set when the job was triggered by a workflow.
	// It allows correlating job execution with workflow runs.
	WorkflowContext *WorkflowContext `json:"workflow_context,omitempty"`
}

JobInfo contains information about a platform job.

func (*JobInfo) HasWorkflowContext

func (j *JobInfo) HasWorkflowContext() bool

HasWorkflowContext returns true if the job was triggered by a workflow.

type JobPoller

type JobPoller struct {
	// contains filtered or unexported fields
}

JobPoller polls for and executes platform jobs.

The JobPoller uses long-polling to efficiently wait for jobs. When no jobs are available, the server holds the connection open until a job arrives or the timeout expires. This provides near-real-time job dispatch with minimal network overhead.

Optional integrations:

  • ResourceController: Throttle job acceptance based on CPU/memory
  • AuditLogger: Log job lifecycle events for debugging and compliance

func NewJobPoller

func NewJobPoller(client JobClient, executor JobExecutor, config *PollerConfig) *JobPoller

NewJobPoller creates a new JobPoller.

func (*JobPoller) CurrentJobCount

func (p *JobPoller) CurrentJobCount() int

CurrentJobCount returns the current number of running jobs.

func (*JobPoller) SetAuditLogger

func (p *JobPoller) SetAuditLogger(logger AuditLogger)

SetAuditLogger sets the optional audit logger for event logging. When set, job lifecycle events will be logged.

func (*JobPoller) SetLeaseManager

func (p *JobPoller) SetLeaseManager(lm *LeaseManager)

SetLeaseManager sets the lease manager for job count reporting. It also registers a callback to cancel all running jobs when lease expires.

func (*JobPoller) SetResourceController

func (p *JobPoller) SetResourceController(rc ResourceController)

SetResourceController sets the optional resource controller for throttling. When set, jobs will only be accepted when resources are available.

func (*JobPoller) Start

func (p *JobPoller) Start(ctx context.Context) error

Start starts the job poller.

func (*JobPoller) Stop

func (p *JobPoller) Stop(timeout time.Duration) error

Stop stops the job poller and waits for active jobs to complete.

type JobResult

type JobResult struct {
	JobID         string                 `json:"job_id"`
	Status        string                 `json:"status"` // completed, failed, canceled
	CompletedAt   time.Time              `json:"completed_at"`
	DurationMs    int64                  `json:"duration_ms"`
	FindingsCount int                    `json:"findings_count"`
	Error         string                 `json:"error,omitempty"`
	Metadata      map[string]interface{} `json:"metadata,omitempty"`

	// WorkflowContext is echoed back if the job was triggered by a workflow.
	// This allows the Workflow Executor to correlate results with workflow runs.
	WorkflowContext *WorkflowContext `json:"workflow_context,omitempty"`
}

JobResult contains the result of a completed job.

func (*JobResult) HasWorkflowContext

func (r *JobResult) HasWorkflowContext() bool

HasWorkflowContext returns true if the result includes workflow context.

type JobValidationError

type JobValidationError struct {
	JobID  string
	Reason string
}

JobValidationError represents a job validation failure.

func (*JobValidationError) Error

func (e *JobValidationError) Error() string

type KeyChoice added in v0.12.0

type KeyChoice struct {
	// APIKey is the key to use.
	APIKey string
	// ExpiresAt is its expiry when known (pass it to
	// KeyRenewConfig.CurrentKeyExpiresAt).
	ExpiresAt *time.Time
	// NeverExpires is true when the server said the key has no expiry (pass
	// it to KeyRenewConfig.CurrentKeyNeverExpires). False when unknown.
	NeverExpires bool
	// FromStateFile is true when the key is a renewed key read from the
	// credentials file rather than the configured one.
	FromStateFile bool
	// Reason explains the choice, for the sensor's log.
	Reason string
}

KeyChoice is the API key a sensor starts with.

func ChooseAPIKey added in v0.12.0

func ChooseAPIKey(store *FileCredentialStore, configuredKey, sensorID string) KeyChoice

ChooseAPIKey picks the key a sensor starts with, from its configured key (SENSOR_API_KEY, a flag, a config file) and the credentials file. The renewed key in the file wins, because a renewal retires the configured key, except:

  • the file is missing, unreadable or empty, or belongs to another sensor (sensorID set and different): the configured key;
  • the configured key is not the one the file's key descends from (an administrator regenerated the key and the operator configured the new one): the configured key. The file is replaced on the next renewal.

A file written before fingerprints existed has no fingerprint; its key wins, as it did before. With no configured key the file's key is used.

type KeyRenewConfig added in v0.5.0

type KeyRenewConfig struct {
	// RenewFraction is the fraction of the key lifetime at which to renew
	// (0 < f < 1). Default 0.5.
	RenewFraction float64

	// MinInterval floors the computed schedule. Default 1m.
	MinInterval time.Duration

	// RetryInterval is the wait before retrying a failed renewal. Default 30s.
	RetryInterval time.Duration

	// OnRotated is called after a successful key swap so the caller can persist
	// the new key (e.g. FileCredentialStore.Save) for the next restart.
	//
	// A non-nil error is treated as a failure, not a warning: the server has
	// already invalidated the previous key, so an unpersisted new key means
	// the next restart comes up with a dead credential. The manager reports
	// the error (stderr + OnPersistError), keeps the running client on the new
	// key, and retries the persist every RetryInterval. It does NOT rotate
	// again until the persist succeeds — rotating on top of an unsaved key
	// would only widen the gap.
	OnRotated func(newKey string, expiresAt *time.Time) error

	// OnPersistError, if set, is called each time OnRotated fails. Use it to
	// surface the condition (metrics, health check, alert).
	OnPersistError func(err error)

	// CurrentKeyExpiresAt is the expiry of the key the client starts with,
	// when the caller knows it (e.g. stored next to the key). When set, the
	// first renewal is scheduled from it instead of rotating immediately.
	CurrentKeyExpiresAt *time.Time

	// CurrentKeyNeverExpires tells the manager the server issued the current
	// key without a TTL. The manager then does nothing: there is nothing to
	// renew, and the discovery rotation would replace a working key for no
	// benefit. Leave false when unknown — the manager then performs one
	// discovery renewal to learn the server's TTL posture.
	CurrentKeyNeverExpires bool

	// Verbose enables debug logging.
	Verbose bool
}

KeyRenewConfig configures the KeyRenewManager.

type KeyRenewManager added in v0.5.0

type KeyRenewManager struct {
	// contains filtered or unexported fields
}

KeyRenewManager auto-renews the sensor's API key before it expires, then swaps it into the live client and persists it. It is a no-op when the server has no key TTL (renewal returns a nil expiry) — after the discovery renewal it stops.

Rotation model: this pairs with a server that issues short-lived keys. On the current server the previous key is invalidated the instant renewal succeeds (no overlap window yet), so a request already in flight at the swap instant can see a 401; the poll and lease loops self-heal on their next tick. Zero- window overlap is a planned server-side change.

func NewKeyRenewManager added in v0.5.0

func NewKeyRenewManager(client KeyRenewer, config *KeyRenewConfig) *KeyRenewManager

NewKeyRenewManager creates a KeyRenewManager. A nil config uses defaults.

func (*KeyRenewManager) RenewNow added in v0.7.1

func (m *KeyRenewManager) RenewNow()

RenewNow asks the running manager to renew the key now instead of at its schedule, e.g. because the platform's heartbeat said the key is inside its renewal window (the rotate_key doorbell action). It never blocks; requests made while one is pending collapse into it, and a request within MinInterval of the last rotation is ignored (the platform judged the key that was just replaced).

func (*KeyRenewManager) Start added in v0.5.0

func (m *KeyRenewManager) Start(ctx context.Context) error

Start launches the renewal loop in the background. Without a known expiry it performs one discovery renewal immediately: if the server returns no expiry (TTL disabled), nothing more is scheduled and the sensor keeps its non-expiring key. Otherwise it schedules the next renewal at RenewFraction of the remaining lifetime and repeats. In every case RenewNow still renews on request while the loop runs.

func (*KeyRenewManager) Stop added in v0.5.0

func (m *KeyRenewManager) Stop()

Stop stops the renewal loop.

type KeyRenewer added in v0.5.0

type KeyRenewer interface {
	RenewKey(ctx context.Context) (*RenewKeyResponse, error)
	SetAPIKey(key string)
}

KeyRenewer is the subset of PlatformClient the renew manager needs: fetch a fresh key and swap it in. *PlatformClient satisfies this.

type LeaseClient

type LeaseClient interface {
	RenewLease(ctx context.Context, req *LeaseRenewRequest) (*LeaseRenewResponse, error)
	ReleaseLease(ctx context.Context) error
}

LeaseClient defines the interface for lease operations.

func NewHTTPLeaseClient

func NewHTTPLeaseClient(baseURL, apiKey, sensorID string) LeaseClient

NewHTTPLeaseClient creates a new HTTP-based lease client.

type LeaseConfig

type LeaseConfig struct {
	// LeaseDuration is how long the lease is valid for.
	// Default: 60 seconds.
	LeaseDuration time.Duration

	// RenewInterval is how often to renew the lease.
	// Should be less than LeaseDuration (typically 1/3).
	// Default: 20 seconds.
	RenewInterval time.Duration

	// GracePeriod is how long to wait after lease expiry before considering sensor dead.
	// Default: 15 seconds.
	GracePeriod time.Duration

	// MaxJobs is the maximum concurrent jobs this sensor can handle.
	MaxJobs int

	// MetricsCollector provides system metrics for lease renewal.
	// If nil, metrics are not reported.
	MetricsCollector MetricsCollector

	// OnLeaseExpired is called when the lease expires (sensor should shutdown).
	OnLeaseExpired func()

	// Verbose enables debug logging.
	Verbose bool

	// UseSecureIdentity enables cryptographically secure holder identity.
	// When true, the holder identity includes a random nonce that cannot be guessed.
	// Default: true (should only be false for testing).
	UseSecureIdentity *bool

	// IdentityPrefix is an optional prefix for the holder identity.
	// Useful for identifying sensor type (e.g., "scanner", "collector").
	IdentityPrefix string
}

LeaseConfig configures the LeaseManager.

type LeaseInfo

type LeaseInfo struct {
	SensorID         string    `json:"agent_id"`
	HolderIdentity   string    `json:"holder_identity"`
	LeaseDurationSec int       `json:"lease_duration_seconds"`
	AcquireTime      time.Time `json:"acquire_time"`
	RenewTime        time.Time `json:"renew_time"`
	ResourceVersion  int       `json:"resource_version"`
}

LeaseInfo contains information about the current lease.

type LeaseManager

type LeaseManager struct {
	// contains filtered or unexported fields
}

LeaseManager manages the sensor's lease with the control plane. It periodically renews the lease to indicate the sensor is healthy.

func NewLeaseManager

func NewLeaseManager(client LeaseClient, config *LeaseConfig) *LeaseManager

NewLeaseManager creates a new LeaseManager.

func (*LeaseManager) DecrementJobs

func (m *LeaseManager) DecrementJobs(failed bool)

DecrementJobs decrements the current job count by 1 and increments completed.

func (*LeaseManager) GetStatus

func (m *LeaseManager) GetStatus() *LeaseStatus

GetStatus returns the current lease status.

func (*LeaseManager) IncrementJobs

func (m *LeaseManager) IncrementJobs()

IncrementJobs increments the current job count by 1.

func (*LeaseManager) SetCurrentJobs

func (m *LeaseManager) SetCurrentJobs(count int)

SetCurrentJobs updates the current number of jobs being processed.

func (*LeaseManager) Start

func (m *LeaseManager) Start(ctx context.Context) error

Start starts the lease renewal loop.

func (*LeaseManager) Stop

func (m *LeaseManager) Stop(ctx context.Context) error

Stop stops the lease manager and releases the lease.

type LeaseRenewRequest

type LeaseRenewRequest struct {
	HolderIdentity       string   `json:"holder_identity"`
	LeaseDurationSeconds int      `json:"lease_duration_seconds"`
	CurrentJobs          int      `json:"current_jobs"`
	MaxJobs              int      `json:"max_jobs"`
	CPUPercent           *float64 `json:"cpu_percent,omitempty"`
	MemoryPercent        *float64 `json:"memory_percent,omitempty"`
	DiskPercent          *float64 `json:"disk_percent,omitempty"`
	JobsCompletedTotal   int      `json:"jobs_completed_total,omitempty"`
	JobsFailedTotal      int      `json:"jobs_failed_total,omitempty"`
}

LeaseRenewRequest contains the data for renewing a lease.

type LeaseRenewResponse

type LeaseRenewResponse struct {
	Success         bool      `json:"success"`
	Message         string    `json:"message,omitempty"`
	ResourceVersion int       `json:"resource_version"`
	RenewTime       time.Time `json:"renew_time"`
}

LeaseRenewResponse contains the response from lease renewal.

type LeaseStatus

type LeaseStatus struct {
	Running         bool
	Healthy         bool
	LastRenewTime   time.Time
	ResourceVersion int
	CurrentJobs     int
	LastError       error
}

LeaseStatus represents the current status of the lease.

type MetricsCollector

type MetricsCollector interface {
	Collect() (*SystemMetrics, error)
}

MetricsCollector collects system metrics.

type PlatformClient

type PlatformClient struct {
	// contains filtered or unexported fields
}

PlatformClient provides a unified interface for platform sensor operations. It implements LeaseClient and JobClient interfaces.

func NewPlatformClient

func NewPlatformClient(config *ClientConfig) *PlatformClient

NewPlatformClient creates a new PlatformClient.

func (*PlatformClient) AcknowledgeJob

func (c *PlatformClient) AcknowledgeJob(ctx context.Context, jobID string) error

AcknowledgeJob implements JobClient.

func (*PlatformClient) Poll

Poll implements JobClient.

func (*PlatformClient) ReleaseLease

func (c *PlatformClient) ReleaseLease(ctx context.Context) error

ReleaseLease implements LeaseClient.

func (*PlatformClient) RenewKey added in v0.5.0

func (c *PlatformClient) RenewKey(ctx context.Context) (*RenewKeyResponse, error)

RenewKey rotates this sensor's API key by presenting the current one (POST /api/v2/sensor/keys, api RFC-029 §4.7). A platform that does not serve the route (404/405 without a problem document) answers ErrRenewUnsupported: protocol v1 is retired, there is no fall-back. It does NOT swap the key in — the caller decides when to call SetAPIKey (and persist), so a failed persist never leaves the running client on a key the sensor can't recover after a restart. It never retries: without a key TTL the platform replaces the presented key, so a repeated renewal after a lost answer would be refused.

func (*PlatformClient) RenewLease

RenewLease implements LeaseClient.

func (*PlatformClient) ReportJobProgress

func (c *PlatformClient) ReportJobProgress(ctx context.Context, jobID string, progress int, message string) error

ReportJobProgress implements JobClient.

func (*PlatformClient) ReportJobResult

func (c *PlatformClient) ReportJobResult(ctx context.Context, result *JobResult) error

ReportJobResult implements JobClient.

func (*PlatformClient) SetAPIKey added in v0.5.0

func (c *PlatformClient) SetAPIKey(key string)

SetAPIKey atomically swaps the API key used by all subsequent lease and job requests. Safe to call concurrently with in-flight requests.

type PlatformSensor added in v0.7.0

type PlatformSensor struct {
	// contains filtered or unexported fields
}

PlatformSensor represents a fully configured platform sensor.

func (*PlatformSensor) AuditLogger added in v0.7.0

func (a *PlatformSensor) AuditLogger() *audit.Logger

AuditLogger returns the audit logger if configured.

func (*PlatformSensor) ExtendedStatus added in v0.7.0

func (a *PlatformSensor) ExtendedStatus() *SensorStatus

ExtendedStatus returns the full sensor status including resource metrics.

func (*PlatformSensor) FlushPipeline added in v0.7.0

func (a *PlatformSensor) FlushPipeline(ctx context.Context) error

FlushPipeline waits for all pending uploads to complete. Returns an error if the pipeline is not configured or if the context is canceled.

func (*PlatformSensor) Pipeline added in v0.7.0

func (a *PlatformSensor) Pipeline() *pipeline.Pipeline

Pipeline returns the upload pipeline if configured.

func (*PlatformSensor) PipelineStats added in v0.7.0

func (a *PlatformSensor) PipelineStats() *pipeline.Stats

PipelineStats returns the current pipeline statistics. Returns nil if the pipeline is not configured.

func (*PlatformSensor) ResourceController added in v0.7.0

func (a *PlatformSensor) ResourceController() *resource.Controller

ResourceController returns the resource controller if configured.

func (*PlatformSensor) SmartSubmitReport added in v0.7.0

func (a *PlatformSensor) SmartSubmitReport(_ context.Context, report *ctis.Report, opts ...pipeline.SubmitOption) (string, error)

SmartSubmitReport queues a report on the upload pipeline. Large reports need no special path: the client sends them as protocol v2 segments. It returns the pipeline item id, or an error when no pipeline is configured.

func (*PlatformSensor) Start added in v0.7.0

func (a *PlatformSensor) Start(ctx context.Context) error

Start starts the platform sensor (lease manager + job poller).

func (*PlatformSensor) Status added in v0.7.0

func (a *PlatformSensor) Status() *SensorStatus

Status returns the current sensor status.

func (*PlatformSensor) Stop added in v0.7.0

func (a *PlatformSensor) Stop(ctx context.Context, timeout time.Duration) error

Stop stops the platform sensor gracefully.

func (*PlatformSensor) SubmitReport added in v0.7.0

func (a *PlatformSensor) SubmitReport(report *ctis.Report, opts ...pipeline.SubmitOption) (string, error)

SubmitReport queues a report for async upload via the pipeline. Returns immediately after queueing. Use Pipeline().GetStats() to monitor progress. Returns an error if the pipeline is not configured.

type PollRequest

type PollRequest struct {
	MaxJobs        int      `json:"max_jobs"`
	Capabilities   []string `json:"capabilities,omitempty"`
	TimeoutSeconds int      `json:"timeout_seconds,omitempty"`
}

PollRequest contains the data for polling jobs.

type PollResponse

type PollResponse struct {
	Jobs             []*JobInfo `json:"jobs"`
	PollIntervalHint int        `json:"poll_interval_hint,omitempty"` // Suggested wait before next poll
	QueueDepth       int        `json:"queue_depth,omitempty"`        // Total pending jobs
}

PollResponse contains the response from job polling.

type PollerConfig

type PollerConfig struct {
	// MaxConcurrentJobs is the maximum number of concurrent jobs.
	MaxConcurrentJobs int

	// PollTimeout is the long-poll timeout (how long to wait for jobs).
	PollTimeout time.Duration

	// RetryDelay is the delay between poll attempts on error.
	RetryDelay time.Duration

	// Capabilities to advertise when polling.
	Capabilities []string

	// OnJobStarted is called when a job starts executing.
	OnJobStarted func(job *JobInfo)

	// OnJobCompleted is called when a job completes (success or failure).
	OnJobCompleted func(job *JobInfo, result *JobResult)

	// Verbose enables debug logging.
	Verbose bool

	// AllowedJobTypes restricts which job types can be executed.
	// If empty, all job types are allowed.
	AllowedJobTypes []string

	// MaxPayloadSize limits the maximum payload size for jobs (default: 10MB).
	MaxPayloadSize int

	// RequireAuthToken requires jobs to have a valid auth token.
	// When true, jobs without AuthToken will be rejected.
	RequireAuthToken bool

	// ValidateTokenClaims enables JWT claims validation.
	// When true, the AuthToken's tenant_id claim must match job's TenantID.
	ValidateTokenClaims bool
}

PollerConfig configures the JobPoller.

type RegistrationRequest

type RegistrationRequest struct {
	// Name is the sensor's display name.
	Name string `json:"name"`

	// Capabilities are the scanner/collector capabilities (e.g., "sast", "sca", "dast").
	Capabilities []string `json:"capabilities"`

	// Tools are the specific tools available (e.g., "semgrep", "trivy", "nuclei").
	Tools []string `json:"tools,omitempty"`

	// Region is the deployment region (e.g., "us-east-1", "ap-southeast-1").
	Region string `json:"region,omitempty"`

	// Labels are optional key-value labels for filtering.
	Labels map[string]string `json:"labels,omitempty"`

	// MaxConcurrentJobs is the maximum number of concurrent jobs.
	MaxConcurrentJobs int `json:"max_concurrent_jobs,omitempty"`
}

RegistrationRequest contains the data for registering a new platform sensor.

type RegistrationResponse

type RegistrationResponse struct {
	// SensorID is the pre-rename "agent_id" key (legacyv1.FieldSensorID).
	SensorID  string `json:"agent_id"`
	APIKey    string `json:"api_key"`    // Only returned once - store securely!
	APIPrefix string `json:"api_prefix"` // Prefix for display/logging (safe to log)
	Message   string `json:"message,omitempty"`
}

RegistrationResponse contains the response from sensor registration.

type RenewError added in v0.9.0

type RenewError struct{ StatusCode int }

RenewError is a refused key renewal. It satisfies the HTTPStatusCode interface pkg/core uses to tell a rejected key from other failures.

func (*RenewError) Error added in v0.9.0

func (e *RenewError) Error() string

func (*RenewError) HTTPStatusCode added in v0.9.0

func (e *RenewError) HTTPStatusCode() int

HTTPStatusCode returns the response status.

type RenewKeyResponse added in v0.5.0

type RenewKeyResponse struct {
	APIKey    string     `json:"api_key"`
	ExpiresAt *time.Time `json:"expires_at,omitempty"`
}

RenewKeyResponse is the result of a self-renew call. ExpiresAt is nil when the server has no key TTL configured (the key never expires).

type ResourceController

type ResourceController interface {
	// AcquireSlot attempts to acquire a job execution slot.
	// Returns true if acquired, false if throttled or at capacity.
	AcquireSlot(ctx context.Context) bool

	// ReleaseSlot releases a previously acquired slot.
	ReleaseSlot()

	// IsThrottled returns true if resource limits are exceeded.
	IsThrottled() bool
}

ResourceController is an optional interface for resource-based job throttling. If set on JobPoller, jobs will only be accepted when resources are available.

type SensorBuilder added in v0.7.0

type SensorBuilder struct {
	// contains filtered or unexported fields
}

SensorBuilder provides a fluent API for building a platform sensor.

func NewSensorBuilder added in v0.7.0

func NewSensorBuilder() *SensorBuilder

NewSensorBuilder creates a new SensorBuilder.

func (*SensorBuilder) Build added in v0.7.0

func (b *SensorBuilder) Build() (*PlatformSensor, error)

Build creates a PlatformSensor from the builder configuration.

func (*SensorBuilder) OnJobCompleted added in v0.7.0

func (b *SensorBuilder) OnJobCompleted(fn func(*JobInfo, *JobResult)) *SensorBuilder

OnJobCompleted sets the callback for job completion.

func (*SensorBuilder) OnJobStarted added in v0.7.0

func (b *SensorBuilder) OnJobStarted(fn func(*JobInfo)) *SensorBuilder

OnJobStarted sets the callback for job start.

func (*SensorBuilder) OnLeaseExpired added in v0.7.0

func (b *SensorBuilder) OnLeaseExpired(fn func()) *SensorBuilder

OnLeaseExpired sets the callback for lease expiration.

func (*SensorBuilder) WithAuditLogger added in v0.7.0

func (b *SensorBuilder) WithAuditLogger(config *audit.LoggerConfig) *SensorBuilder

WithAuditLogger enables audit logging with the given config. When enabled, all job lifecycle events will be logged.

func (*SensorBuilder) WithCapabilities added in v0.7.0

func (b *SensorBuilder) WithCapabilities(caps ...string) *SensorBuilder

WithCapabilities sets the sensor capabilities.

func (*SensorBuilder) WithCredentials added in v0.7.0

func (b *SensorBuilder) WithCredentials(baseURL, apiKey, sensorID string) *SensorBuilder

WithCredentials sets the sensor credentials.

func (*SensorBuilder) WithExecutor added in v0.7.0

func (b *SensorBuilder) WithExecutor(executor JobExecutor) *SensorBuilder

WithExecutor sets the job executor.

func (*SensorBuilder) WithLeaseDuration added in v0.7.0

func (b *SensorBuilder) WithLeaseDuration(d time.Duration) *SensorBuilder

WithLeaseDuration sets the lease duration.

func (*SensorBuilder) WithMaxJobs added in v0.7.0

func (b *SensorBuilder) WithMaxJobs(n int) *SensorBuilder

WithMaxJobs sets the maximum concurrent jobs.

func (*SensorBuilder) WithMetricsCollector added in v0.7.0

func (b *SensorBuilder) WithMetricsCollector(collector MetricsCollector) *SensorBuilder

WithMetricsCollector sets the metrics collector.

func (*SensorBuilder) WithPipeline added in v0.7.0

func (b *SensorBuilder) WithPipeline(config *pipeline.PipelineConfig, uploader pipeline.Uploader) *SensorBuilder

WithPipeline enables the async upload pipeline. The pipeline allows scan results to be uploaded asynchronously in the background, so scans can complete immediately without waiting for uploads.

func (*SensorBuilder) WithPollTimeout added in v0.7.0

func (b *SensorBuilder) WithPollTimeout(d time.Duration) *SensorBuilder

WithPollTimeout sets the poll timeout.

func (*SensorBuilder) WithRenewInterval added in v0.7.0

func (b *SensorBuilder) WithRenewInterval(d time.Duration) *SensorBuilder

WithRenewInterval sets the lease renewal interval.

func (*SensorBuilder) WithResourceController added in v0.7.0

func (b *SensorBuilder) WithResourceController(config *resource.ControllerConfig) *SensorBuilder

WithResourceController enables resource throttling with the given config. When enabled, jobs will only be accepted when CPU/memory are below thresholds.

func (*SensorBuilder) WithVerbose added in v0.7.0

func (b *SensorBuilder) WithVerbose(v bool) *SensorBuilder

WithVerbose enables verbose logging.

type SensorCredentials added in v0.7.0

type SensorCredentials struct {
	SensorID  string `json:"sensor_id"`
	APIKey    string `json:"api_key"`
	APIPrefix string `json:"api_prefix"`
	// ExpiresAt is the key's expiry when known (from a key renewal). Pass it
	// to KeyRenewConfig.CurrentKeyExpiresAt on startup so the renewer does
	// not rotate a still-valid key immediately.
	ExpiresAt *time.Time `json:"expires_at,omitempty"`
	// ConfiguredKeyFingerprint is the salted PBKDF2 fingerprint
	// (KeyFingerprint) of the configured key APIKey was renewed from. When
	// the configured key no longer matches it, an administrator issued a new
	// key and the operator configured it: ChooseAPIKey then uses the
	// configured key. Empty in files written before it existed.
	ConfiguredKeyFingerprint string `json:"configured_key_fingerprint,omitempty"`
	// NeverExpires records that the server issued APIKey without an expiry
	// (its key TTL is off), as opposed to an expiry that is not known. Pass it
	// to KeyRenewConfig.CurrentKeyNeverExpires so a restart does not rotate
	// the key again just to learn that.
	NeverExpires bool `json:"never_expires,omitempty"`
}

SensorCredentials contains the credentials returned after sensor registration.

This is also the format of the credentials file (FileCredentialStore). The sensor id is written as "sensor_id"; files written before the agent -> sensor rename carry "agent_id" and are still read (see UnmarshalJSON).

func EnsureRegistered

func EnsureRegistered(ctx context.Context, config *EnsureRegisteredConfig) (*SensorCredentials, error)

EnsureRegistered ensures the sensor is registered, either by loading existing credentials or registering with a bootstrap token.

This is the recommended way to initialize a platform sensor:

creds, err := platform.EnsureRegistered(ctx, &platform.EnsureRegisteredConfig{
    BaseURL: "http://localhost:8080",
    BootstrapToken: os.Getenv("BOOTSTRAP_TOKEN"),
    CredentialsFile: "/etc/openctem/credentials.json",
    Registration: &platform.RegistrationRequest{
        Name: "scanner-001",
        Capabilities: []string{"sast", "sca"},
    },
})

With an empty CredentialsFile the default ~/.openctem/sensor-credentials.json is used, and a credentials file left by a sensor from before the agent -> sensor rename (~/.openctem/agent-credentials.json) is moved there first, so the sensor keeps its identity and key and does not register again (see ResolveCredentialsFile). An explicit path is used as is.

No server for this flow: no OpenCTEM API serves POST /api/v1/platform/register (bootstrap tokens were never built; api RFC-032 §3.4). Sensors authenticate with the API key an administrator issues; enrollment tokens (api RFC-032 Phase 2) replace this flow. Kept so code that compiles against it keeps compiling, until it is removed.

func (*SensorCredentials) UnmarshalJSON added in v0.7.0

func (c *SensorCredentials) UnmarshalJSON(data []byte) error

UnmarshalJSON reads both the current "sensor_id" key and the "agent_id" key credentials files written before the agent -> sensor rename carry. Both present with different values is an error: the file has been edited by hand and the sensor's identity is ambiguous.

type SensorInfo added in v0.7.0

type SensorInfo struct {
	ID           string    `json:"id"`
	Name         string    `json:"name"`
	Capabilities []string  `json:"capabilities"`
	Region       string    `json:"region"`
	Status       string    `json:"status"`
	CreatedAt    time.Time `json:"created_at"`
}

SensorInfo contains information about a registered platform sensor.

type SensorStatus added in v0.7.0

type SensorStatus struct {
	SensorID    string
	Running     bool
	Healthy     bool
	CurrentJobs int
	LastRenew   time.Time
	LastError   error

	// Resource status (if controller is enabled)
	ResourceStatus *resource.ControllerStatus
}

SensorStatus represents the current sensor status.

type SimpleMetricsCollector

type SimpleMetricsCollector struct{}

SimpleMetricsCollector provides basic system metrics collection.

func (*SimpleMetricsCollector) Collect

func (c *SimpleMetricsCollector) Collect() (*SystemMetrics, error)

Collect collects system metrics. This is a simplified implementation - production should use proper system calls.

type StatePersistence added in v0.12.0

type StatePersistence struct {
	Persistent bool
	// Reason explains the verdict, for the sensor's log.
	Reason string
}

StatePersistence says whether a state directory survives the sensor being recreated.

func CheckStatePersistence added in v0.12.0

func CheckStatePersistence(dir string) StatePersistence

CheckStatePersistence reports whether dir survives the sensor being recreated. Outside a container the host filesystem persists. Inside one, dir must be on a mounted volume (the nearest mount at or above dir: a volume at /var/lib/openctem covers /var/lib/openctem/state) (a Docker volume, a bind mount, a Kubernetes volume) that is not a tmpfs: the container's writable layer is lost when the container is recreated. A Kubernetes emptyDir looks like a volume and is lost with the pod; the Helm chart therefore decides key renewal explicitly instead of relying on this check.

type SystemMetrics

type SystemMetrics struct {
	CPUPercent    float64 `json:"cpu_percent"`
	MemoryPercent float64 `json:"memory_percent"`
	DiskPercent   float64 `json:"disk_percent"`
	CurrentJobs   int     `json:"current_jobs"`
	MaxJobs       int     `json:"max_jobs"`
}

SystemMetrics contains sensor system metrics for health reporting.

type WorkflowContext

type WorkflowContext struct {
	// WorkflowID is the UUID of the workflow definition.
	WorkflowID string `json:"workflow_id"`

	// WorkflowRunID is the UUID of the specific workflow execution.
	WorkflowRunID string `json:"workflow_run_id"`

	// TriggerType indicates what triggered the workflow (e.g., "finding_created", "schedule", "manual").
	TriggerType string `json:"trigger_type"`

	// ActionNodeID is the UUID of the action node that triggered this job.
	ActionNodeID string `json:"action_node_id"`

	// ActionNodeKey is the node_key of the action node (e.g., "run_scan_1").
	ActionNodeKey string `json:"action_node_key"`
}

WorkflowContext contains context information when a job is triggered by the Workflow Executor as part of an automation workflow.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL