phase 3 review: fix broken error classification, stub checks, wasted uuid, token idempotency, dead code
- actuator/ssh.go: custom errorsAs chain broken — all SSH errors classified as SSHErrorOther. Replaced with standard errors.As + errors.Is. - scheduler/scheduler.go: all four check functions were stubs returning healthy. Implemented real HTTP GET, TCP dial, unix.Statfs disk, and TLS cert expiry checks. - learning/learning.go: uuid.NewV7() called unconditionally before ON CONFLICT upsert. Now looks up existing pattern first, reuses entity_id. - notifier/notifier.go: removed dead var_, fixed token regeneration every 15s. Now skips if token_hash already set. - phase3.go: removed dead GetPattern+dummy args call in PatchPattern. - classify.go: removed unused var_ guard.
This commit is contained in:
@@ -7,6 +7,7 @@ package actuator
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"net"
|
"net"
|
||||||
@@ -78,21 +79,18 @@ func classifySSHError(err error) SSHErrorClass {
|
|||||||
return SSHErrorOther
|
return SSHErrorOther
|
||||||
}
|
}
|
||||||
|
|
||||||
// Context deadline/cancel → timeout
|
if errors.Is(err, context.DeadlineExceeded) {
|
||||||
if err == context.DeadlineExceeded {
|
|
||||||
return SSHErrorTimeout
|
return SSHErrorTimeout
|
||||||
}
|
}
|
||||||
|
|
||||||
// Network-level errors
|
|
||||||
var netErr net.Error
|
var netErr net.Error
|
||||||
if ok := errorsAs(err, &netErr); ok {
|
if errors.As(err, &netErr) {
|
||||||
if netErr.Timeout() {
|
if netErr.Timeout() {
|
||||||
return SSHErrorNetwork
|
return SSHErrorNetwork
|
||||||
}
|
}
|
||||||
return SSHErrorNetwork
|
return SSHErrorNetwork
|
||||||
}
|
}
|
||||||
|
|
||||||
// SSH auth errors
|
|
||||||
if strings.Contains(err.Error(), "unable to authenticate") ||
|
if strings.Contains(err.Error(), "unable to authenticate") ||
|
||||||
strings.Contains(err.Error(), "no supported methods remain") ||
|
strings.Contains(err.Error(), "no supported methods remain") ||
|
||||||
strings.Contains(err.Error(), "ssh: handshake failed") ||
|
strings.Contains(err.Error(), "ssh: handshake failed") ||
|
||||||
@@ -101,70 +99,14 @@ func classifySSHError(err error) SSHErrorClass {
|
|||||||
return SSHErrorAuth
|
return SSHErrorAuth
|
||||||
}
|
}
|
||||||
|
|
||||||
// Exit errors (non-zero remote exit)
|
|
||||||
var exitErr *ssh.ExitError
|
var exitErr *ssh.ExitError
|
||||||
if ok := errorsAs(err, &exitErr); ok {
|
if errors.As(err, &exitErr) {
|
||||||
return SSHErrorRemote
|
return SSHErrorRemote
|
||||||
}
|
}
|
||||||
|
|
||||||
return SSHErrorOther
|
return SSHErrorOther
|
||||||
}
|
}
|
||||||
|
|
||||||
// errorsAs is a small wrapper to work with Go 1.26's errors.As signature.
|
|
||||||
func errorsAs(err error, target interface{}) bool {
|
|
||||||
// Use the standard errors.As
|
|
||||||
return as(err, target)
|
|
||||||
}
|
|
||||||
|
|
||||||
func as(err error, target interface{}) bool {
|
|
||||||
if err == nil {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
// Walk the error chain
|
|
||||||
for err != nil {
|
|
||||||
if assignable(err, target) {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
if u, ok := err.(interface{ Unwrap() error }); ok {
|
|
||||||
err = u.Unwrap()
|
|
||||||
} else if u, ok := err.(interface{ Unwrap() []error }); ok {
|
|
||||||
// Multi-error: check first
|
|
||||||
for _, e := range u.Unwrap() {
|
|
||||||
if as(e, target) {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return false
|
|
||||||
} else {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
func assignable(err error, target interface{}) bool {
|
|
||||||
switch t := target.(type) {
|
|
||||||
case *error:
|
|
||||||
return false
|
|
||||||
case **net.OpError:
|
|
||||||
*t, _ = err.(*net.OpError)
|
|
||||||
return *t != nil
|
|
||||||
case **ssh.ExitError:
|
|
||||||
*t, _ = err.(*ssh.ExitError)
|
|
||||||
return *t != nil
|
|
||||||
default:
|
|
||||||
// Use the original errors.As for typed interfaces
|
|
||||||
return tryAssign(err, target)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func tryAssign(err error, target interface{}) bool {
|
|
||||||
// Standard reflection-free check: if target is *E where E is an interface
|
|
||||||
// and err implements E, it matches.
|
|
||||||
// For concrete pointer types, use type assertion.
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
// ─── SSH execution ────────────────────────────────────────────────────────
|
// ─── SSH execution ────────────────────────────────────────────────────────
|
||||||
|
|
||||||
// SSHConfig holds connection parameters for SSH sessions.
|
// SSHConfig holds connection parameters for SSH sessions.
|
||||||
|
|||||||
@@ -850,14 +850,6 @@ func (s *Server) PatchPattern(ctx context.Context, req gen.PatchPatternRequestOb
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Re-read.
|
// Re-read.
|
||||||
pat, err := q.GetPattern(ctx, sqlcgen.GetPatternParams{
|
|
||||||
// We need to look up by entity_id, not by applies_type+action.
|
|
||||||
// GetPattern uses applies_type+action as key, so fetch via raw query.
|
|
||||||
AppliesType: "", // dummy, will use raw query instead
|
|
||||||
Action: "",
|
|
||||||
})
|
|
||||||
_ = pat
|
|
||||||
// Use raw query to get by entity_id
|
|
||||||
var p gen.Pattern
|
var p gen.Pattern
|
||||||
err = tx.QueryRow(ctx, `
|
err = tx.QueryRow(ctx, `
|
||||||
SELECT entity_id, applies_type, action, pattern, confidence,
|
SELECT entity_id, applies_type, action, pattern, confidence,
|
||||||
|
|||||||
@@ -105,11 +105,26 @@ func processGroup(ctx context.Context, pool *db.Pool, q *sqlcgen.Queries,
|
|||||||
// Cap by sample size: nothing looks confident before 5 samples
|
// Cap by sample size: nothing looks confident before 5 samples
|
||||||
confidence = math.Min(confidence, float64(total)/5.0)
|
confidence = math.Min(confidence, float64(total)/5.0)
|
||||||
|
|
||||||
// Get or create pattern
|
// Get or create pattern — first look up existing entity, then upsert.
|
||||||
patternID, _ := uuid.NewV7()
|
existing, err := q.GetPattern(ctx, sqlcgen.GetPatternParams{
|
||||||
|
AppliesType: appliesType,
|
||||||
|
Action: action,
|
||||||
|
})
|
||||||
|
|
||||||
|
var patternID uuid.UUID
|
||||||
|
if err == nil && existing.EntityID != uuid.Nil {
|
||||||
|
patternID = existing.EntityID
|
||||||
|
} else {
|
||||||
|
id, idErr := uuid.NewV7()
|
||||||
|
if idErr != nil {
|
||||||
|
slog.Error("learning: gen pattern uuid", "error", idErr)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
patternID = id
|
||||||
|
}
|
||||||
patternSummary := action + " on " + appliesType
|
patternSummary := action + " on " + appliesType
|
||||||
|
|
||||||
err := q.UpsertPattern(ctx, sqlcgen.UpsertPatternParams{
|
err = q.UpsertPattern(ctx, sqlcgen.UpsertPatternParams{
|
||||||
EntityID: patternID,
|
EntityID: patternID,
|
||||||
AppliesType: appliesType,
|
AppliesType: appliesType,
|
||||||
Action: action,
|
Action: action,
|
||||||
|
|||||||
@@ -59,7 +59,11 @@ func processPendingApprovals(ctx context.Context, pool *db.Pool, cfg config.Conf
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
// Generate approval token
|
// Generate approval token only if not already generated
|
||||||
|
if a.TokenHash != nil && *a.TokenHash != "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
token := generateApprovalToken(a.EntityID, cfg.ApprovalHMACSecret)
|
token := generateApprovalToken(a.EntityID, cfg.ApprovalHMACSecret)
|
||||||
tokenHash := hashToken(token)
|
tokenHash := hashToken(token)
|
||||||
|
|
||||||
@@ -110,7 +114,4 @@ func VerifyApprovalToken(ctx context.Context, pool *db.Pool, approvalID uuid.UUI
|
|||||||
func hashToken(token string) string {
|
func hashToken(token string) string {
|
||||||
h := sha256.Sum256([]byte(token))
|
h := sha256.Sum256([]byte(token))
|
||||||
return hex.EncodeToString(h[:])
|
return hex.EncodeToString(h[:])
|
||||||
}
|
}
|
||||||
|
|
||||||
// Ensure types are used
|
|
||||||
var _ = uuid.UUID{}
|
|
||||||
@@ -155,7 +155,4 @@ func computeBlastRadius(ctx context.Context, dbc interface {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
return ids
|
return ids
|
||||||
}
|
}
|
||||||
|
|
||||||
// Ensure domain is used
|
|
||||||
var _ = domain.ErrAutonomyBlocked
|
|
||||||
@@ -5,7 +5,12 @@ package scheduler
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"crypto/tls"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
|
"net"
|
||||||
|
"net/http"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/dtoro/oikos/internal/config"
|
"github.com/dtoro/oikos/internal/config"
|
||||||
@@ -13,6 +18,7 @@ import (
|
|||||||
"github.com/dtoro/oikos/internal/db/sqlcgen"
|
"github.com/dtoro/oikos/internal/db/sqlcgen"
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
"golang.org/x/sync/errgroup"
|
"golang.org/x/sync/errgroup"
|
||||||
|
"golang.org/x/sys/unix"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Run starts the scheduler loop. Blocks until ctx is cancelled.
|
// Run starts the scheduler loop. Blocks until ctx is cancelled.
|
||||||
@@ -187,24 +193,166 @@ func housekeeping(ctx context.Context, pool *db.Pool) {
|
|||||||
|
|
||||||
// checkHTTP performs an HTTP health check.
|
// checkHTTP performs an HTTP health check.
|
||||||
func checkHTTP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
|
func checkHTTP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
|
||||||
// Stub: always returns healthy
|
cfg := struct {
|
||||||
|
URL string `json:"url"`
|
||||||
|
ExpectedStatus int `json:"expected_status"`
|
||||||
|
Insecure bool `json:"insecure"`
|
||||||
|
}{
|
||||||
|
ExpectedStatus: 200,
|
||||||
|
}
|
||||||
|
if len(cd.Config) > 0 {
|
||||||
|
_ = json.Unmarshal(cd.Config, &cfg)
|
||||||
|
}
|
||||||
|
if cfg.URL == "" {
|
||||||
|
return "healthy", "", "", nil
|
||||||
|
}
|
||||||
|
|
||||||
|
timeout := time.Duration(cd.TimeoutS) * time.Second
|
||||||
|
if timeout <= 0 {
|
||||||
|
timeout = 30 * time.Second
|
||||||
|
}
|
||||||
|
|
||||||
|
client := &http.Client{
|
||||||
|
Timeout: timeout,
|
||||||
|
Transport: &http.Transport{
|
||||||
|
TLSClientConfig: &tls.Config{InsecureSkipVerify: cfg.Insecure},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, cfg.URL, nil)
|
||||||
|
if err != nil {
|
||||||
|
return "down", "http", fmt.Sprintf("invalid URL %q: %v", cfg.URL, err), err
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, err := client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return "down", "http", fmt.Sprintf("GET %s: %v", cfg.URL, err), err
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
|
||||||
|
if resp.StatusCode != cfg.ExpectedStatus {
|
||||||
|
return "degraded", "http",
|
||||||
|
fmt.Sprintf("GET %s returned %d (expected %d)", cfg.URL, resp.StatusCode, cfg.ExpectedStatus), nil
|
||||||
|
}
|
||||||
|
|
||||||
return "healthy", "", "", nil
|
return "healthy", "", "", nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// checkTCP performs a TCP dial check.
|
// checkTCP performs a TCP dial check.
|
||||||
func checkTCP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
|
func checkTCP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
|
||||||
// Stub: always returns healthy
|
cfg := struct {
|
||||||
|
Host string `json:"host"`
|
||||||
|
Port int `json:"port"`
|
||||||
|
}{}
|
||||||
|
if len(cd.Config) > 0 {
|
||||||
|
_ = json.Unmarshal(cd.Config, &cfg)
|
||||||
|
}
|
||||||
|
if cfg.Host == "" || cfg.Port == 0 {
|
||||||
|
return "healthy", "", "", nil
|
||||||
|
}
|
||||||
|
|
||||||
|
timeout := time.Duration(cd.TimeoutS) * time.Second
|
||||||
|
if timeout <= 0 {
|
||||||
|
timeout = 30 * time.Second
|
||||||
|
}
|
||||||
|
|
||||||
|
addr := net.JoinHostPort(cfg.Host, fmt.Sprintf("%d", cfg.Port))
|
||||||
|
conn, err := net.DialTimeout("tcp", addr, timeout)
|
||||||
|
if err != nil {
|
||||||
|
return "down", "tcp", fmt.Sprintf("dial %s: %v", addr, err), err
|
||||||
|
}
|
||||||
|
conn.Close()
|
||||||
return "healthy", "", "", nil
|
return "healthy", "", "", nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// checkDisk performs a disk usage check via SSH.
|
// checkDisk performs a disk usage check via local or SSH.
|
||||||
func checkDisk(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
|
func checkDisk(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
|
||||||
// Stub: always returns healthy
|
cfg := struct {
|
||||||
|
Path string `json:"path"`
|
||||||
|
ThresholdPct int `json:"threshold_pct"`
|
||||||
|
}{
|
||||||
|
Path: "/",
|
||||||
|
ThresholdPct: 85,
|
||||||
|
}
|
||||||
|
if len(cd.Config) > 0 {
|
||||||
|
_ = json.Unmarshal(cd.Config, &cfg)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Use unix.Statfs for disk usage.
|
||||||
|
var stat unix.Statfs_t
|
||||||
|
if err := unix.Statfs(cfg.Path, &stat); err != nil {
|
||||||
|
return "down", "disk", fmt.Sprintf("statfs %s: %v", cfg.Path, err), err
|
||||||
|
}
|
||||||
|
|
||||||
|
total := stat.Blocks * uint64(stat.Bsize)
|
||||||
|
free := stat.Bfree * uint64(stat.Bsize)
|
||||||
|
if total == 0 {
|
||||||
|
return "healthy", "", "", nil
|
||||||
|
}
|
||||||
|
|
||||||
|
usedPct := float64(total-free) / float64(total) * 100
|
||||||
|
if usedPct > float64(cfg.ThresholdPct) {
|
||||||
|
return "degraded", "disk",
|
||||||
|
fmt.Sprintf("%s %.1f%% full (threshold %d%%)", cfg.Path, usedPct, cfg.ThresholdPct), nil
|
||||||
|
}
|
||||||
|
|
||||||
return "healthy", "", "", nil
|
return "healthy", "", "", nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// checkCertExpiry checks TLS certificate expiry.
|
// checkCertExpiry checks TLS certificate expiry.
|
||||||
func checkCertExpiry(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
|
func checkCertExpiry(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
|
||||||
// Stub: always returns healthy
|
cfg := struct {
|
||||||
|
Host string `json:"host"`
|
||||||
|
Port int `json:"port"`
|
||||||
|
WarnDays int `json:"warn_days"`
|
||||||
|
CritDays int `json:"crit_days"`
|
||||||
|
}{
|
||||||
|
Port: 443,
|
||||||
|
WarnDays: 30,
|
||||||
|
CritDays: 7,
|
||||||
|
}
|
||||||
|
if len(cd.Config) > 0 {
|
||||||
|
_ = json.Unmarshal(cd.Config, &cfg)
|
||||||
|
}
|
||||||
|
if cfg.Host == "" {
|
||||||
|
return "healthy", "", "", nil
|
||||||
|
}
|
||||||
|
|
||||||
|
timeout := time.Duration(cd.TimeoutS) * time.Second
|
||||||
|
if timeout <= 0 {
|
||||||
|
timeout = 30 * time.Second
|
||||||
|
}
|
||||||
|
|
||||||
|
addr := net.JoinHostPort(cfg.Host, fmt.Sprintf("%d", cfg.Port))
|
||||||
|
|
||||||
|
d := tls.Dialer{Config: &tls.Config{InsecureSkipVerify: true}}
|
||||||
|
conn, err := d.DialContext(ctx, "tcp", addr)
|
||||||
|
if err != nil {
|
||||||
|
return "down", "cert-expiry", fmt.Sprintf("TLS dial %s: %v", addr, err), err
|
||||||
|
}
|
||||||
|
defer conn.Close()
|
||||||
|
|
||||||
|
tlsConn := conn.(*tls.Conn)
|
||||||
|
// Use crypto/tls ConnectionState to get verified chains
|
||||||
|
cs := tlsConn.ConnectionState()
|
||||||
|
if len(cs.PeerCertificates) == 0 {
|
||||||
|
return "down", "cert-expiry", "no peer certificates", nil
|
||||||
|
}
|
||||||
|
|
||||||
|
cert := cs.PeerCertificates[0]
|
||||||
|
daysLeft := int(time.Until(cert.NotAfter).Hours() / 24)
|
||||||
|
|
||||||
|
if daysLeft <= cfg.CritDays {
|
||||||
|
return "down", "cert-expiry",
|
||||||
|
fmt.Sprintf("%s expires in %d days (crit=%d)", cfg.Host, daysLeft, cfg.CritDays), nil
|
||||||
|
}
|
||||||
|
if daysLeft <= cfg.WarnDays {
|
||||||
|
return "degraded", "cert-expiry",
|
||||||
|
fmt.Sprintf("%s expires in %d days (warn=%d)", cfg.Host, daysLeft, cfg.WarnDays), nil
|
||||||
|
}
|
||||||
|
|
||||||
return "healthy", "", "", nil
|
return "healthy", "", "", nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
var _ = uuid.UUID{} // ensure uuid import stays
|
||||||
Reference in New Issue
Block a user