You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
719 lines
28 KiB
719 lines
28 KiB
|
2 months ago
|
package main
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"errors"
|
||
|
|
"fmt"
|
||
|
|
"log"
|
||
|
|
"net"
|
||
|
|
"net/http"
|
||
|
|
"os"
|
||
|
|
"os/signal"
|
||
|
|
"strconv"
|
||
|
|
"strings"
|
||
|
|
"sync"
|
||
|
|
"sync/atomic"
|
||
|
|
"syscall"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"tera-clone/services/tera-api/internal/adapters/mock"
|
||
|
|
"tera-clone/services/tera-api/internal/alarmdelivery"
|
||
|
|
"tera-clone/services/tera-api/internal/alarmdelivery/alarmsocket"
|
||
|
|
"tera-clone/services/tera-api/internal/api"
|
||
|
|
"tera-clone/services/tera-api/internal/diagnostics"
|
||
|
|
"tera-clone/services/tera-api/internal/directstream"
|
||
|
|
"tera-clone/services/tera-api/internal/domain"
|
||
|
|
"tera-clone/services/tera-api/internal/hardware/greenledsocket"
|
||
|
|
"tera-clone/services/tera-api/internal/hardware/systemcontrol"
|
||
|
|
"tera-clone/services/tera-api/internal/logruntime"
|
||
|
|
"tera-clone/services/tera-api/internal/logservice"
|
||
|
|
"tera-clone/services/tera-api/internal/managementaccess"
|
||
|
|
"tera-clone/services/tera-api/internal/networkconfig"
|
||
|
|
"tera-clone/services/tera-api/internal/networkconfig/netsocket"
|
||
|
|
"tera-clone/services/tera-api/internal/portservice"
|
||
|
|
"tera-clone/services/tera-api/internal/security/password"
|
||
|
|
"tera-clone/services/tera-api/internal/serial"
|
||
|
|
"tera-clone/services/tera-api/internal/sessionhealth"
|
||
|
|
"tera-clone/services/tera-api/internal/snmpagent"
|
||
|
|
"tera-clone/services/tera-api/internal/snmpagent/snmpsocket"
|
||
|
|
"tera-clone/services/tera-api/internal/storage/atomicfile"
|
||
|
|
"tera-clone/services/tera-api/internal/supportbundle"
|
||
|
|
"tera-clone/services/tera-api/internal/systemmetrics"
|
||
|
|
"tera-clone/services/tera-api/internal/systemtime"
|
||
|
|
"tera-clone/services/tera-api/internal/systemtime/timesocket"
|
||
|
|
"tera-clone/services/tera-api/internal/tlscert"
|
||
|
|
"tera-clone/services/tera-api/internal/update"
|
||
|
|
"tera-clone/services/tera-api/internal/update/updatesocket"
|
||
|
|
)
|
||
|
|
|
||
|
|
func main() {
|
||
|
|
addr := listenAddrConfig()
|
||
|
|
readHeaderTimeout := getenvDurationSeconds("TERA_API_READ_HEADER_TIMEOUT_SECONDS", 5*time.Second)
|
||
|
|
readTimeout := getenvDurationSeconds("TERA_API_READ_TIMEOUT_SECONDS", 15*time.Second)
|
||
|
|
writeTimeout := getenvDurationSeconds("TERA_API_WRITE_TIMEOUT_SECONDS", 30*time.Second)
|
||
|
|
idleTimeout := getenvDurationSeconds("TERA_API_IDLE_TIMEOUT_SECONDS", 60*time.Second)
|
||
|
|
auditPath := getenv("TERA_API_AUDIT_LOG", "")
|
||
|
|
auditMax := getenvInt("TERA_API_AUDIT_MAX_EVENTS", 1000)
|
||
|
|
systemLogPath := getenv("TERA_API_SYSTEM_LOG_STORE", "")
|
||
|
|
portLogPath := getenv("TERA_API_PORT_LOG_STORE", "")
|
||
|
|
linkLogPath := getenv("TERA_API_LINK_LOG_STORE", "")
|
||
|
|
logMaxEvents := getenvInt("TERA_API_LOG_MAX_EVENTS", 10000)
|
||
|
|
authMaxSessions := getenvInt("TERA_API_AUTH_MAX_SESSIONS", 128)
|
||
|
|
authMaxTokens := getenvInt("TERA_API_AUTH_MAX_TOKENS", 128)
|
||
|
|
authMaxLoginFailures := getenvInt("TERA_API_AUTH_MAX_LOGIN_FAILURES", 1024)
|
||
|
|
authMaxUsers := getenvInt("TERA_API_AUTH_MAX_USERS", 64)
|
||
|
|
minimumPasswordLength := getenvInt("TERA_API_MIN_PASSWORD_LENGTH", 8)
|
||
|
|
configPath := getenv("TERA_API_CONFIG_STORE", "")
|
||
|
|
operationPath := getenv("TERA_API_OPERATION_STORE", "")
|
||
|
|
operationMax := getenvInt("TERA_API_OPERATION_MAX_EVENTS", 256)
|
||
|
|
authPath := getenv("TERA_API_AUTH_STORE", "")
|
||
|
|
sessionPath := getenv("TERA_API_SESSION_STORE", "")
|
||
|
|
loginFailurePath := getenv("TERA_API_LOGIN_FAILURE_STORE", "")
|
||
|
|
secretPath := getenv("TERA_API_SECRET_STORE", "/var/lib/tera-api/secrets.json")
|
||
|
|
greenLEDSocket := getenv("TERA_API_GREEN_LED_SOCKET", greenledsocket.DefaultSocketPath)
|
||
|
|
corsOrigins := getenvList("TERA_API_CORS_ORIGINS", nil)
|
||
|
|
maxRequestBytes := int64(getenvInt("TERA_API_MAX_REQUEST_BYTES", 1<<20))
|
||
|
|
allowMockActors := allowMockActorsConfig()
|
||
|
|
diagnosticsRunner, err := diagnosticsRunnerConfig(getenvBool("TERA_API_DIAGNOSTICS_ENABLED", false))
|
||
|
|
if err != nil {
|
||
|
|
log.Fatalf("tera-api diagnostics setup failed: %v", err)
|
||
|
|
}
|
||
|
|
supportBundleService, err := supportBundleServiceConfig(os.Getenv("TERA_API_SUPPORT_BUNDLE_DIR"))
|
||
|
|
if err != nil {
|
||
|
|
log.Fatalf("tera-api support bundle setup failed: %v", err)
|
||
|
|
}
|
||
|
|
alarmDeliveryController := alarmdelivery.Controller(alarmdelivery.UnavailableController{Reason: "TERA_API_ALARM_SOCKET is not configured"})
|
||
|
|
if alarmSocket := os.Getenv("TERA_API_ALARM_SOCKET"); alarmSocket != "" {
|
||
|
|
alarmDeliveryController = alarmsocket.NewClient(alarmSocket)
|
||
|
|
} else if allowMockActors {
|
||
|
|
alarmDeliveryController = alarmdelivery.MockController{}
|
||
|
|
}
|
||
|
|
timeController := systemtime.Controller(systemtime.UnavailableController{Reason: "TERA_API_TIME_SOCKET is not configured"})
|
||
|
|
if timeSocket := os.Getenv("TERA_API_TIME_SOCKET"); timeSocket != "" {
|
||
|
|
timeController = timesocket.NewClient(timeSocket)
|
||
|
|
} else if allowMockActors {
|
||
|
|
timeController = systemtime.MockController{}
|
||
|
|
}
|
||
|
|
networkController := networkconfig.Controller(networkconfig.UnavailableController{Reason: "TERA_API_NETWORK_SOCKET is not configured"})
|
||
|
|
if networkSocket := os.Getenv("TERA_API_NETWORK_SOCKET"); networkSocket != "" {
|
||
|
|
networkController = netsocket.NewClient(networkSocket)
|
||
|
|
} else if allowMockActors {
|
||
|
|
networkController = &networkconfig.MockController{}
|
||
|
|
}
|
||
|
|
managementController := managementaccess.Controller(managementaccess.NoopController{})
|
||
|
|
if managementSocket := os.Getenv("TERA_API_MANAGEMENT_SOCKET"); managementSocket != "" {
|
||
|
|
managementController = managementaccess.NewSocketClient(managementSocket)
|
||
|
|
}
|
||
|
|
snmpAgentController := snmpagent.Controller(snmpagent.UnavailableController{Reason: "TERA_API_SNMP_SOCKET is not configured"})
|
||
|
|
if snmpSocket := os.Getenv("TERA_API_SNMP_SOCKET"); snmpSocket != "" {
|
||
|
|
snmpAgentController = snmpsocket.NewClient(snmpSocket)
|
||
|
|
} else if allowMockActors {
|
||
|
|
snmpAgentController = &snmpagent.MockController{}
|
||
|
|
}
|
||
|
|
serialController := serial.Controller(serial.UnavailableController{Reason: "TERA_API_SERIAL_SOCKET is not configured"})
|
||
|
|
if serialSocket := os.Getenv("TERA_API_SERIAL_SOCKET"); serialSocket != "" {
|
||
|
|
serialController = serial.NewSocketClient(serialSocket)
|
||
|
|
} else if allowMockActors {
|
||
|
|
serialController = serial.MockController{}
|
||
|
|
}
|
||
|
|
sessionController := sessionhealth.Controller(sessionhealth.UnavailableController{Reason: "TERA_API_SESSIOND_STATUS_URL is not configured"})
|
||
|
|
if sessionStatusURL := os.Getenv("TERA_API_SESSIOND_STATUS_URL"); sessionStatusURL != "" {
|
||
|
|
client, sessionErr := sessionhealth.NewHTTPClient(sessionStatusURL)
|
||
|
|
if sessionErr != nil {
|
||
|
|
log.Fatalf("tera-api sessiond status setup failed: %v", sessionErr)
|
||
|
|
}
|
||
|
|
sessionController = client
|
||
|
|
}
|
||
|
|
firmwareController := update.Controller(update.UnavailableController{Reason: "TERA_API_UPDATE_SOCKET is not configured"})
|
||
|
|
if updateSocket := os.Getenv("TERA_API_UPDATE_SOCKET"); updateSocket != "" {
|
||
|
|
firmwareController = updatesocket.NewClient(updateSocket)
|
||
|
|
}
|
||
|
|
systemController := systemcontrol.Controller(systemcontrol.UnavailableController{Reason: "TERA_API_SYSTEM_SOCKET is not configured"})
|
||
|
|
if systemSocket := os.Getenv("TERA_API_SYSTEM_SOCKET"); systemSocket != "" {
|
||
|
|
systemController = systemcontrol.NewClient(systemSocket)
|
||
|
|
} else if allowMockActors {
|
||
|
|
systemController = systemcontrol.MockController{}
|
||
|
|
}
|
||
|
|
passwordKDF, passwordKDFs, err := passwordKDFConfig(getenv("TERA_API_PASSWORD_KDF", password.AlgorithmArgon2id))
|
||
|
|
if err != nil {
|
||
|
|
log.Fatalf("tera-api password KDF setup failed: %v", err)
|
||
|
|
}
|
||
|
|
var logJournal logservice.Journal
|
||
|
|
if logSocket := os.Getenv("TERA_API_LOG_SOCKET"); logSocket != "" {
|
||
|
|
logJournal = logservice.NewClient(logSocket)
|
||
|
|
}
|
||
|
|
tlsCertificates, err := tlscert.Open(tlscert.Options{
|
||
|
|
Root: getenv("TERA_API_TLS_STORE", "/var/lib/tera-api/tls/library"),
|
||
|
|
ActiveCert: getenv("TERA_API_TLS_ACTIVE_CERT", "/var/lib/tera-api/tls/active.crt"),
|
||
|
|
ActiveKey: getenv("TERA_API_TLS_ACTIVE_KEY", "/var/lib/tera-api/tls/active.key"),
|
||
|
|
Clock: time.Now,
|
||
|
|
})
|
||
|
|
if err != nil {
|
||
|
|
log.Fatalf("tera-api TLS certificate store setup failed: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
store, err := mock.NewStoreWithOptions(time.Now, mock.StoreOptions{
|
||
|
|
LogJournal: logJournal,
|
||
|
|
AllowSampleLogs: allowMockActors,
|
||
|
|
AuditPath: auditPath,
|
||
|
|
AuditMaxEvents: auditMax,
|
||
|
|
SystemLogPath: systemLogPath,
|
||
|
|
SystemLogMaxEvents: logMaxEvents,
|
||
|
|
PortLogPath: portLogPath,
|
||
|
|
PortLogMaxEvents: logMaxEvents,
|
||
|
|
LinkLogPath: linkLogPath,
|
||
|
|
LinkLogMaxEvents: logMaxEvents,
|
||
|
|
ConfigPath: configPath,
|
||
|
|
OperationPath: operationPath,
|
||
|
|
OperationMaxEvents: operationMax,
|
||
|
|
AuthPath: authPath,
|
||
|
|
SessionPath: sessionPath,
|
||
|
|
LoginFailurePath: loginFailurePath,
|
||
|
|
SecretPath: secretPath,
|
||
|
|
AuthMaxSessions: authMaxSessions,
|
||
|
|
AuthMaxTokens: authMaxTokens,
|
||
|
|
AuthMaxLoginFailures: authMaxLoginFailures,
|
||
|
|
AuthMaxUsers: authMaxUsers,
|
||
|
|
MinimumPasswordLength: minimumPasswordLength,
|
||
|
|
PasswordKDF: passwordKDF,
|
||
|
|
PasswordKDFs: passwordKDFs,
|
||
|
|
TimeController: timeController,
|
||
|
|
NetworkController: networkController,
|
||
|
|
ManagementController: managementController,
|
||
|
|
SystemController: systemController,
|
||
|
|
AlarmDeliveryController: alarmDeliveryController,
|
||
|
|
SNMPAgentController: snmpAgentController,
|
||
|
|
TLSCertificateStore: tlsCertificates,
|
||
|
|
DiagnosticsRunner: diagnosticsRunner,
|
||
|
|
SupportBundleService: supportBundleService,
|
||
|
|
DiagnosticsWorkers: getenvInt("TERA_API_DIAGNOSTICS_WORKERS", 2),
|
||
|
|
DiagnosticsQueue: getenvInt("TERA_API_DIAGNOSTICS_QUEUE", 16),
|
||
|
|
RemoveLegacyAuthUsers: true,
|
||
|
|
})
|
||
|
|
if err != nil {
|
||
|
|
log.Fatalf("tera-api store setup failed: %v", err)
|
||
|
|
}
|
||
|
|
defer store.Close()
|
||
|
|
startupTasks := make([]func(), 0, 6)
|
||
|
|
bootIDPath := getenv("TERA_API_BOOT_ID_PATH", "/proc/sys/kernel/random/boot_id")
|
||
|
|
bootMarkerPath := getenv("TERA_API_BOOT_MARKER_PATH", "/var/lib/tera-api/last-notified-boot-id")
|
||
|
|
startupTasks = append(startupTasks, func() {
|
||
|
|
if bootID, unnotified, bootErr := unnotifiedKernelBoot(bootIDPath, bootMarkerPath); bootErr != nil {
|
||
|
|
log.Printf("tera-api cold-start detection degraded: %v", bootErr)
|
||
|
|
} else if unnotified {
|
||
|
|
lifecycleContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||
|
|
notifyErr := store.EmitLifecycleAlarm(lifecycleContext, "cold_start", "device cold start detected")
|
||
|
|
cancel()
|
||
|
|
if notifyErr != nil {
|
||
|
|
log.Printf("tera-api cold-start alarm degraded: %v", notifyErr)
|
||
|
|
} else if markerErr := atomicfile.Write(bootMarkerPath, []byte(bootID+"\n"), 0o600); markerErr != nil {
|
||
|
|
log.Printf("tera-api cold-start marker degraded: %v", markerErr)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
})
|
||
|
|
startupTasks = append(startupTasks, func() {
|
||
|
|
reconcileContext, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
reconcileErr := store.ReconcileTLSRuntime(reconcileContext)
|
||
|
|
if reconcileErr != nil {
|
||
|
|
log.Printf("tera-api TLS certificate reconciliation degraded: %v", reconcileErr)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
startupTasks = append(startupTasks, func() {
|
||
|
|
reconcileContext, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
reconcileErr := store.ReconcileSNMPAgentRuntime(reconcileContext)
|
||
|
|
if reconcileErr != nil {
|
||
|
|
log.Printf("tera-api SNMP agent reconciliation degraded: %v", reconcileErr)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
var portServiceController portservice.Controller = portservice.NoopController{}
|
||
|
|
if getenvBool("TERA_API_PORT_SERVICES_ENABLED", !allowMockActors) {
|
||
|
|
signer, signerErr := portservice.LoadOrCreateSSHSigner(getenv("TERA_API_PORT_SSH_HOST_KEY", "/var/lib/tera-api/port-service-ed25519"))
|
||
|
|
if signerErr != nil {
|
||
|
|
log.Fatalf("tera-api channel SSH host key setup failed: %v", signerErr)
|
||
|
|
}
|
||
|
|
streamFactory := serial.NewStreamSocketClient(getenv("TERA_API_SERIAL_STREAM_SOCKET", serial.DefaultStreamSocketPath), "tera-port-service")
|
||
|
|
gateway, gatewayErr := portservice.New(portservice.Options{
|
||
|
|
TransportFactory: streamFactory,
|
||
|
|
ListenHost: getenv("TERA_API_PORT_LISTEN_HOST", "0.0.0.0"),
|
||
|
|
SSHSigner: signer,
|
||
|
|
KnownHostsPath: getenv("TERA_API_PORT_SSH_KNOWN_HOSTS", "/etc/tera/ssh_known_hosts"),
|
||
|
|
AuthenticateMethod: func(ctx context.Context, method, username, credential, channel string) (string, error) {
|
||
|
|
return store.AuthenticatePort(ctx, method, username, credential, channel)
|
||
|
|
},
|
||
|
|
Record: func(ctx context.Context, event domain.PortLogEvent) error {
|
||
|
|
_, recordErr := store.RecordPortLog(ctx, event)
|
||
|
|
return recordErr
|
||
|
|
},
|
||
|
|
})
|
||
|
|
if gatewayErr != nil {
|
||
|
|
log.Fatalf("tera-api channel operation service setup failed: %v", gatewayErr)
|
||
|
|
}
|
||
|
|
defer gateway.Close()
|
||
|
|
portServiceController = gateway
|
||
|
|
startupTasks = append(startupTasks, func() {
|
||
|
|
reconcileContext, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
snapshot, snapshotErr := store.ConfigExport(reconcileContext)
|
||
|
|
if snapshotErr != nil {
|
||
|
|
log.Printf("tera-api channel operation reconciliation degraded: %v", snapshotErr)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
servicePorts, credentialErr := store.ResolvePortServiceCredentials(reconcileContext, "running", snapshot.Running.Ports)
|
||
|
|
if credentialErr != nil {
|
||
|
|
log.Printf("tera-api channel operation credential reconciliation degraded: %v", credentialErr)
|
||
|
|
} else if applyErr := gateway.Apply(reconcileContext, servicePorts); applyErr != nil {
|
||
|
|
log.Printf("tera-api channel operation reconciliation degraded: %v", applyErr)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
}
|
||
|
|
startupTasks = append(startupTasks, func() {
|
||
|
|
reconcileContext, cancel := context.WithTimeout(context.Background(), 12*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
reconcileErr := reconcileManagementRuntime(reconcileContext, store, managementController)
|
||
|
|
if reconcileErr != nil {
|
||
|
|
log.Printf("tera-api management access reconciliation degraded: %v", reconcileErr)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
startupTasks = append(startupTasks, func() {
|
||
|
|
reconcileContext, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
reconcileErr := reconcileSerialRuntime(reconcileContext, store, serialController)
|
||
|
|
if reconcileErr != nil {
|
||
|
|
log.Printf("tera-api serial runtime reconciliation degraded: %v", reconcileErr)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
logRuntimeContext, stopLogRuntime := context.WithCancel(context.Background())
|
||
|
|
defer stopLogRuntime()
|
||
|
|
if resetButton, ok := systemController.(resetButtonStateReader); ok {
|
||
|
|
go runFactoryResetButtonMonitor(
|
||
|
|
logRuntimeContext,
|
||
|
|
resetButton,
|
||
|
|
store,
|
||
|
|
getenvDurationMilliseconds("TERA_API_RESET_BUTTON_POLL_MILLISECONDS", 100*time.Millisecond),
|
||
|
|
getenvDurationMilliseconds("TERA_API_RESET_BUTTON_HOLD_MILLISECONDS", time.Second),
|
||
|
|
func(operation domain.Operation) {
|
||
|
|
log.Printf("tera-api hardware settings reset completed: %s", operation.OperationID)
|
||
|
|
},
|
||
|
|
func(err error) { log.Printf("tera-api hardware settings reset monitor degraded: %v", err) },
|
||
|
|
)
|
||
|
|
}
|
||
|
|
go runSerialRuntimeReconciler(
|
||
|
|
logRuntimeContext, store, serialController,
|
||
|
|
getenvDurationSeconds("TERA_API_SERIAL_RECONCILE_INTERVAL_SECONDS", 30*time.Second),
|
||
|
|
func(err error) {
|
||
|
|
if err != nil {
|
||
|
|
log.Printf("tera-api serial runtime reconciliation degraded: %v", err)
|
||
|
|
} else {
|
||
|
|
log.Printf("tera-api serial runtime reconciliation restored")
|
||
|
|
}
|
||
|
|
},
|
||
|
|
)
|
||
|
|
if logJournal != nil {
|
||
|
|
go runOperationalAlarmEvaluator(
|
||
|
|
logRuntimeContext, store,
|
||
|
|
getenvDurationSeconds("TERA_API_OPERATIONAL_ALARM_INTERVAL_SECONDS", time.Minute),
|
||
|
|
func(err error) { log.Printf("tera-api operational alarm evaluation degraded: %v", err) },
|
||
|
|
)
|
||
|
|
}
|
||
|
|
if getenvBool("TERA_API_LOG_SCHEDULER_ENABLED", logJournal != nil) {
|
||
|
|
interfaces, err := logruntime.ParseEthernetInterfaces(os.Getenv("TERA_API_LOG_ETHERNET_INTERFACES"))
|
||
|
|
if err != nil {
|
||
|
|
log.Fatalf("tera-api log observer setup failed: %v", err)
|
||
|
|
}
|
||
|
|
observers := logruntime.CompositeObserver{
|
||
|
|
logruntime.EthernetObserver{Interfaces: interfaces},
|
||
|
|
}
|
||
|
|
linkController, _ := serialController.(serial.LinkStateController)
|
||
|
|
probeController, _ := serialController.(serial.CRProbeStateController)
|
||
|
|
if linkController != nil || probeController != nil {
|
||
|
|
observers = append(observers, logruntime.SerialObserver{
|
||
|
|
Controller: linkController,
|
||
|
|
Probes: probeController,
|
||
|
|
})
|
||
|
|
}
|
||
|
|
scheduler, err := logruntime.NewScheduler(logruntime.Options{
|
||
|
|
Backend: store, Observer: observers,
|
||
|
|
OnError: func(err error) { log.Printf("tera-api log scheduler degraded: %v", err) },
|
||
|
|
})
|
||
|
|
if err != nil {
|
||
|
|
log.Fatalf("tera-api log scheduler setup failed: %v", err)
|
||
|
|
}
|
||
|
|
go scheduler.Run(logRuntimeContext)
|
||
|
|
}
|
||
|
|
var ticketListener net.Listener
|
||
|
|
var ticketSocketPath string
|
||
|
|
var ticketErrors chan error
|
||
|
|
if ticketSocketPath = os.Getenv("TERA_API_DIRECT_TICKET_SOCKET"); ticketSocketPath != "" {
|
||
|
|
listener, err := directstream.ListenTicketSocket(ticketSocketPath)
|
||
|
|
if err != nil {
|
||
|
|
log.Fatalf("tera-api direct stream ticket socket setup failed: %v", err)
|
||
|
|
}
|
||
|
|
ticketListener = listener
|
||
|
|
defer ticketListener.Close()
|
||
|
|
defer os.Remove(ticketSocketPath)
|
||
|
|
ticketServer := &directstream.TicketSocketServer{Authority: mock.NewDirectTicketAuthority(store)}
|
||
|
|
ticketErrors = make(chan error, 1)
|
||
|
|
go func() { ticketErrors <- ticketServer.Serve(listener) }()
|
||
|
|
log.Printf("tera-api serving Direct Mode tickets on %s", ticketSocketPath)
|
||
|
|
}
|
||
|
|
var startupReady atomic.Bool
|
||
|
|
handler := api.NewRouterWithOptions(store, api.RouterOptions{
|
||
|
|
CORSAllowedOrigins: corsOrigins,
|
||
|
|
MaxRequestBytes: maxRequestBytes,
|
||
|
|
DisableMockActors: !allowMockActors,
|
||
|
|
GreenLEDController: greenledsocket.NewClient(greenLEDSocket),
|
||
|
|
SerialController: serialController,
|
||
|
|
PortServiceController: portServiceController,
|
||
|
|
SessionController: sessionController,
|
||
|
|
NetworkController: networkController,
|
||
|
|
FirmwareController: firmwareController,
|
||
|
|
SystemMetricsReader: systemmetrics.NewLinuxReader(systemmetrics.LinuxReaderOptions{}),
|
||
|
|
MaxFirmwareBytes: int64(getenvInt("TERA_API_UPDATE_MAX_BYTES", 1280<<20)),
|
||
|
|
FirmwareTimeout: getenvDurationSeconds("TERA_API_FIRMWARE_TIMEOUT_SECONDS", 30*time.Minute),
|
||
|
|
CommandsReady: startupReady.Load,
|
||
|
|
})
|
||
|
|
|
||
|
|
server := &http.Server{
|
||
|
|
Addr: addr,
|
||
|
|
Handler: handler,
|
||
|
|
ReadHeaderTimeout: readHeaderTimeout,
|
||
|
|
ReadTimeout: readTimeout,
|
||
|
|
WriteTimeout: writeTimeout,
|
||
|
|
IdleTimeout: idleTimeout,
|
||
|
|
}
|
||
|
|
|
||
|
|
log.Printf("tera-api listening on http://%s", addr)
|
||
|
|
serverErrors := make(chan error, 1)
|
||
|
|
go func() { serverErrors <- server.ListenAndServe() }()
|
||
|
|
go func() {
|
||
|
|
runStartupTasks(startupTasks)
|
||
|
|
startupReady.Store(true)
|
||
|
|
log.Printf("tera-api runtime synchronization completed")
|
||
|
|
}()
|
||
|
|
signalContext, stopSignals := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||
|
|
defer stopSignals()
|
||
|
|
select {
|
||
|
|
case <-signalContext.Done():
|
||
|
|
log.Printf("tera-api stopping")
|
||
|
|
case err := <-serverErrors:
|
||
|
|
if err != nil && err != http.ErrServerClosed {
|
||
|
|
log.Printf("tera-api failed: %v", err)
|
||
|
|
}
|
||
|
|
return
|
||
|
|
case err := <-ticketErrors:
|
||
|
|
if err != nil {
|
||
|
|
log.Printf("tera-api direct stream ticket service failed: %v", err)
|
||
|
|
} else {
|
||
|
|
log.Printf("tera-api direct stream ticket service stopped unexpectedly")
|
||
|
|
}
|
||
|
|
return
|
||
|
|
}
|
||
|
|
if ticketListener != nil {
|
||
|
|
_ = ticketListener.Close()
|
||
|
|
}
|
||
|
|
shutdownContext, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
if err := server.Shutdown(shutdownContext); err != nil {
|
||
|
|
log.Printf("tera-api HTTP shutdown failed: %v", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func runStartupTasks(tasks []func()) {
|
||
|
|
var wait sync.WaitGroup
|
||
|
|
wait.Add(len(tasks))
|
||
|
|
for _, task := range tasks {
|
||
|
|
task := task
|
||
|
|
go func() {
|
||
|
|
defer wait.Done()
|
||
|
|
task()
|
||
|
|
}()
|
||
|
|
}
|
||
|
|
wait.Wait()
|
||
|
|
}
|
||
|
|
|
||
|
|
func unnotifiedKernelBoot(bootIDPath, markerPath string) (string, bool, error) {
|
||
|
|
bootContent, err := os.ReadFile(bootIDPath)
|
||
|
|
if err != nil {
|
||
|
|
return "", false, fmt.Errorf("read kernel boot id: %w", err)
|
||
|
|
}
|
||
|
|
bootID := strings.TrimSpace(string(bootContent))
|
||
|
|
if bootID == "" || len(bootID) > 128 || strings.ContainsAny(bootID, "\r\n\x00") {
|
||
|
|
return "", false, errors.New("kernel boot id is invalid")
|
||
|
|
}
|
||
|
|
markerContent, err := os.ReadFile(markerPath)
|
||
|
|
if err == nil {
|
||
|
|
return bootID, strings.TrimSpace(string(markerContent)) != bootID, nil
|
||
|
|
}
|
||
|
|
if !os.IsNotExist(err) {
|
||
|
|
return "", false, fmt.Errorf("read kernel boot marker: %w", err)
|
||
|
|
}
|
||
|
|
return bootID, true, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
type serialRuntimeConfigStore interface {
|
||
|
|
ConfigExport(context.Context) (domain.ConfigExport, error)
|
||
|
|
}
|
||
|
|
|
||
|
|
type operationalAlarmEvaluator interface {
|
||
|
|
EvaluateOperationalAlarms(context.Context) error
|
||
|
|
}
|
||
|
|
|
||
|
|
func runOperationalAlarmEvaluator(ctx context.Context, evaluator operationalAlarmEvaluator, interval time.Duration, onError func(error)) {
|
||
|
|
if evaluator == nil {
|
||
|
|
return
|
||
|
|
}
|
||
|
|
if interval <= 0 {
|
||
|
|
interval = time.Minute
|
||
|
|
}
|
||
|
|
evaluate := func() {
|
||
|
|
evaluationContext, cancel := context.WithTimeout(ctx, min(interval, 30*time.Second))
|
||
|
|
defer cancel()
|
||
|
|
if err := evaluator.EvaluateOperationalAlarms(evaluationContext); err != nil && onError != nil {
|
||
|
|
onError(err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
evaluate()
|
||
|
|
ticker := time.NewTicker(interval)
|
||
|
|
defer ticker.Stop()
|
||
|
|
for {
|
||
|
|
select {
|
||
|
|
case <-ctx.Done():
|
||
|
|
return
|
||
|
|
case <-ticker.C:
|
||
|
|
evaluate()
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func reconcileManagementRuntime(ctx context.Context, store serialRuntimeConfigStore, controller managementaccess.Controller) error {
|
||
|
|
snapshot, err := store.ConfigExport(ctx)
|
||
|
|
if err != nil {
|
||
|
|
return fmt.Errorf("load running management configuration: %w", err)
|
||
|
|
}
|
||
|
|
if err := controller.Apply(ctx, snapshot.Running.Management); err != nil {
|
||
|
|
return fmt.Errorf("apply running management configuration: %w", err)
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// reconcileSerialRuntime makes persisted line settings real after every
|
||
|
|
// seriald/API restart. seriald remains the sole owner that opens raw TTYs.
|
||
|
|
func reconcileSerialRuntime(ctx context.Context, store serialRuntimeConfigStore, controller serial.Controller) error {
|
||
|
|
snapshot, err := store.ConfigExport(ctx)
|
||
|
|
if err != nil {
|
||
|
|
return fmt.Errorf("load running serial configuration: %w", err)
|
||
|
|
}
|
||
|
|
lineConfigs := make([]serial.ChannelConfig, 0, len(snapshot.Running.Ports))
|
||
|
|
captureConfigs := make([]serial.CaptureConfig, 0, len(snapshot.Running.Ports))
|
||
|
|
for _, port := range snapshot.Running.Ports {
|
||
|
|
lineConfigs = append(lineConfigs, serial.ChannelConfig{
|
||
|
|
Channel: port.Channel, BaudRate: port.BaudRate, DataBits: port.DataBits,
|
||
|
|
Parity: port.Parity, StopBits: port.StopBits, StopBitsMode: port.StopBitsMode,
|
||
|
|
FlowControl: port.FlowControl, RTSControl: port.RTSControl, DTRControl: port.DTRControl,
|
||
|
|
IgnoreJammed: port.IgnoreJammed,
|
||
|
|
})
|
||
|
|
captureConfigs = append(captureConfigs, captureConfigForPort(port))
|
||
|
|
}
|
||
|
|
if len(lineConfigs) != serial.ChannelCount {
|
||
|
|
return fmt.Errorf("running serial configuration contains %d channels, want %d", len(lineConfigs), serial.ChannelCount)
|
||
|
|
}
|
||
|
|
_, lineErr := controller.Apply(ctx, lineConfigs)
|
||
|
|
var captureErr error
|
||
|
|
if captureController, ok := controller.(serial.CaptureController); ok {
|
||
|
|
captureErr = captureController.ApplyCapture(ctx, captureConfigs)
|
||
|
|
}
|
||
|
|
return errors.Join(lineErr, captureErr)
|
||
|
|
}
|
||
|
|
|
||
|
|
func captureConfigForPort(port domain.Port) serial.CaptureConfig {
|
||
|
|
return serial.CaptureConfig{
|
||
|
|
Channel: port.Channel, Mode: port.DataLogMode,
|
||
|
|
DataPackage: serial.CapturePackageConfig{
|
||
|
|
Mode: port.DataPackage.Mode, Length: port.DataPackage.Length, TimeoutMillis: port.DataPackage.TimeoutMillis,
|
||
|
|
DelimiterProcess: port.DataPackage.DelimiterProcess, Delimiter1: port.DataPackage.Delimiter1, Delimiter2: port.DataPackage.Delimiter2,
|
||
|
|
},
|
||
|
|
DataLog: serial.CaptureLogConfig{
|
||
|
|
Read: port.DataLog.Read, Send: port.DataLog.Send, ShowTimestamp: port.DataLog.ShowTimestamp,
|
||
|
|
ShowDirection: port.DataLog.ShowDirection, ShowUsername: port.DataLog.ShowUsername,
|
||
|
|
ShowIPAddress: port.DataLog.ShowIPAddress, ShowPort: port.DataLog.ShowPort, Format: port.DataLog.Format,
|
||
|
|
ReadTrigger: serial.CaptureTriggerConfig{
|
||
|
|
Length: port.DataLog.ReadTrigger.Length, TimeoutMillis: port.DataLog.ReadTrigger.TimeoutMillis,
|
||
|
|
Process: port.DataLog.ReadTrigger.Process, Character1: port.DataLog.ReadTrigger.Character1, Character2: port.DataLog.ReadTrigger.Character2,
|
||
|
|
},
|
||
|
|
SendTrigger: serial.CaptureTriggerConfig{
|
||
|
|
Length: port.DataLog.SendTrigger.Length, TimeoutMillis: port.DataLog.SendTrigger.TimeoutMillis,
|
||
|
|
Process: port.DataLog.SendTrigger.Process, Character1: port.DataLog.SendTrigger.Character1, Character2: port.DataLog.SendTrigger.Character2,
|
||
|
|
},
|
||
|
|
},
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func serialRuntimeHealthy(states []serial.State, err error) bool {
|
||
|
|
if err != nil || len(states) != serial.ChannelCount {
|
||
|
|
return false
|
||
|
|
}
|
||
|
|
for _, state := range states {
|
||
|
|
if !state.Available || !state.ProfileApplied || state.LineSettings == nil {
|
||
|
|
return false
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return true
|
||
|
|
}
|
||
|
|
|
||
|
|
func runSerialRuntimeReconciler(ctx context.Context, store serialRuntimeConfigStore, controller serial.Controller, interval time.Duration, onChange func(error)) {
|
||
|
|
if interval <= 0 {
|
||
|
|
interval = 30 * time.Second
|
||
|
|
}
|
||
|
|
ticker := time.NewTicker(interval)
|
||
|
|
defer ticker.Stop()
|
||
|
|
wasDegraded := false
|
||
|
|
lastError := ""
|
||
|
|
for {
|
||
|
|
select {
|
||
|
|
case <-ctx.Done():
|
||
|
|
return
|
||
|
|
case <-ticker.C:
|
||
|
|
probeContext, cancelProbe := context.WithTimeout(ctx, 3*time.Second)
|
||
|
|
states, stateErr := controller.States(probeContext)
|
||
|
|
cancelProbe()
|
||
|
|
if serialRuntimeHealthy(states, stateErr) {
|
||
|
|
if wasDegraded && onChange != nil {
|
||
|
|
onChange(nil)
|
||
|
|
}
|
||
|
|
wasDegraded = false
|
||
|
|
lastError = ""
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
reconcileContext, cancelReconcile := context.WithTimeout(ctx, 15*time.Second)
|
||
|
|
reconcileErr := reconcileSerialRuntime(reconcileContext, store, controller)
|
||
|
|
cancelReconcile()
|
||
|
|
if reconcileErr == nil {
|
||
|
|
if wasDegraded && onChange != nil {
|
||
|
|
onChange(nil)
|
||
|
|
}
|
||
|
|
wasDegraded = false
|
||
|
|
lastError = ""
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
if !wasDegraded || lastError != reconcileErr.Error() {
|
||
|
|
if onChange != nil {
|
||
|
|
onChange(reconcileErr)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
wasDegraded = true
|
||
|
|
lastError = reconcileErr.Error()
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func listenAddrConfig() string {
|
||
|
|
if value := getenv("TERA_API_ADDR", ""); value != "" {
|
||
|
|
return value
|
||
|
|
}
|
||
|
|
return getenv("TERA_API_LISTEN_ADDR", "127.0.0.1:8080")
|
||
|
|
}
|
||
|
|
|
||
|
|
func getenv(key, fallback string) string {
|
||
|
|
if value := os.Getenv(key); value != "" {
|
||
|
|
return value
|
||
|
|
}
|
||
|
|
return fallback
|
||
|
|
}
|
||
|
|
|
||
|
|
func getenvInt(key string, fallback int) int {
|
||
|
|
value := os.Getenv(key)
|
||
|
|
if value == "" {
|
||
|
|
return fallback
|
||
|
|
}
|
||
|
|
parsed, err := strconv.Atoi(value)
|
||
|
|
if err != nil || parsed <= 0 {
|
||
|
|
return fallback
|
||
|
|
}
|
||
|
|
return parsed
|
||
|
|
}
|
||
|
|
|
||
|
|
func getenvDurationSeconds(key string, fallback time.Duration) time.Duration {
|
||
|
|
value := getenvInt(key, int(fallback/time.Second))
|
||
|
|
return time.Duration(value) * time.Second
|
||
|
|
}
|
||
|
|
|
||
|
|
func getenvDurationMilliseconds(key string, fallback time.Duration) time.Duration {
|
||
|
|
value := getenvInt(key, int(fallback/time.Millisecond))
|
||
|
|
return time.Duration(value) * time.Millisecond
|
||
|
|
}
|
||
|
|
|
||
|
|
func getenvBool(key string, fallback bool) bool {
|
||
|
|
value := strings.TrimSpace(strings.ToLower(os.Getenv(key)))
|
||
|
|
if value == "" {
|
||
|
|
return fallback
|
||
|
|
}
|
||
|
|
switch value {
|
||
|
|
case "1", "true", "yes", "on":
|
||
|
|
return true
|
||
|
|
case "0", "false", "no", "off":
|
||
|
|
return false
|
||
|
|
default:
|
||
|
|
return fallback
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func getenvList(key string, fallback []string) []string {
|
||
|
|
value := os.Getenv(key)
|
||
|
|
if value == "" {
|
||
|
|
return fallback
|
||
|
|
}
|
||
|
|
parts := strings.Split(value, ",")
|
||
|
|
result := make([]string, 0, len(parts))
|
||
|
|
for _, part := range parts {
|
||
|
|
item := strings.TrimSpace(part)
|
||
|
|
if item != "" {
|
||
|
|
result = append(result, item)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if len(result) == 0 {
|
||
|
|
return fallback
|
||
|
|
}
|
||
|
|
return result
|
||
|
|
}
|
||
|
|
|
||
|
|
func allowMockActorsConfig() bool {
|
||
|
|
return getenvBool("TERA_API_ALLOW_MOCK_ACTORS", false)
|
||
|
|
}
|
||
|
|
|
||
|
|
func diagnosticsRunnerConfig(enabled bool) (diagnostics.Runner, error) {
|
||
|
|
if !enabled {
|
||
|
|
return nil, nil
|
||
|
|
}
|
||
|
|
return diagnostics.NewProcessRunner(diagnostics.ProcessRunnerOptions{
|
||
|
|
PingPath: getenv("TERA_API_DIAGNOSTICS_PING_PATH", "/bin/ping"),
|
||
|
|
TraceroutePath: getenv("TERA_API_DIAGNOSTICS_TRACEROUTE_PATH", "/usr/bin/traceroute"),
|
||
|
|
})
|
||
|
|
}
|
||
|
|
|
||
|
|
func supportBundleServiceConfig(directory string) (supportbundle.Service, error) {
|
||
|
|
if strings.TrimSpace(directory) == "" {
|
||
|
|
return nil, nil
|
||
|
|
}
|
||
|
|
return supportbundle.NewManager(supportbundle.Options{
|
||
|
|
Directory: directory,
|
||
|
|
MaxArtifacts: getenvInt("TERA_API_SUPPORT_BUNDLE_MAX_ARTIFACTS", 4),
|
||
|
|
MaxBytes: getenvInt("TERA_API_SUPPORT_BUNDLE_MAX_BYTES", 8*1024*1024),
|
||
|
|
MaxSection: getenvInt("TERA_API_SUPPORT_BUNDLE_MAX_SECTION_BYTES", 2*1024*1024),
|
||
|
|
})
|
||
|
|
}
|
||
|
|
|
||
|
|
func passwordKDFConfig(value string) (password.KDF, password.Registry, error) {
|
||
|
|
mode := strings.TrimSpace(strings.ToLower(value))
|
||
|
|
switch mode {
|
||
|
|
case "", password.AlgorithmArgon2id:
|
||
|
|
return password.NewArgon2id(password.Argon2idDefaultParams()), password.NewProductionRegistry(), nil
|
||
|
|
case password.AlgorithmPBKDF2SHA256, "pbkdf2", "fips":
|
||
|
|
kdf := password.NewPBKDF2SHA256(password.PBKDF2SHA256DefaultParams())
|
||
|
|
return kdf, password.NewRegistry(kdf, password.PrototypeSHA256{}), nil
|
||
|
|
default:
|
||
|
|
return nil, password.Registry{}, fmt.Errorf("unsupported password KDF %q", value)
|
||
|
|
}
|
||
|
|
}
|