marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Made Orborus capable of local compliance and response

Browse Source
This commit is contained in:
Frikky
2026-04-13 23:22:03 +02:00
parent 7a4416d9ce
commit 4d1f3a0ba7
3 changed files with 554 additions and 379 deletions
Download Patch File Download Diff File Expand all files Collapse all files
functions/onprem/orborus/go.mod
+2 -2
View File
@@ -4,7 +4,7 @@ go 1.24.0
toolchain go1.24.4
//replace github.com/shuffle/shuffle-shared => ../../../../shuffle-shared
replace github.com/shuffle/shuffle-shared => ../../../../shuffle-shared
require (
github.com/docker/docker v28.3.3+incompatible
@@ -58,7 +58,7 @@ require (
github.com/envoyproxy/protoc-gen-validate v1.2.1 // indirect
github.com/felixge/httpsnoop v1.0.4 // indirect
github.com/frikky/kin-openapi v0.42.0 // indirect
github.com/frikky/schemaless v0.0.28 // indirect
github.com/frikky/schemaless v0.0.33 // indirect
github.com/fxamacker/cbor/v2 v2.9.0 // indirect
github.com/ghodss/yaml v1.0.0 // indirect
github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect
functions/onprem/orborus/go.sum
+4 -4
View File
@@ -128,8 +128,8 @@ github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2
github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
github.com/frikky/kin-openapi v0.42.0 h1:d5Z6vnuQ6RnCCPIxZaDL+TH2ODLxT8abytOt+Zh+Kd0=
github.com/frikky/kin-openapi v0.42.0/go.mod h1:ev9OZAw7Bv5p0w93j91++6a1ElPzGcCofst+kmrWsj4=
github.com/frikky/schemaless v0.0.32 h1:tbLxdi3GIJZaQDbfiCGEOFDogeBYaHn+IF+Vp9Xzqv4=
github.com/frikky/schemaless v0.0.32/go.mod h1:m9s+6gALXhA5ZERCrJw+jI2rRtTPNa8mkl4vav9sxnY=
github.com/frikky/schemaless v0.0.33 h1:5Soj6VQc+ozqLh4R6MatWOl/atAeNpdon+nV5EKwjOI=
github.com/frikky/schemaless v0.0.33/go.mod h1:m9s+6gALXhA5ZERCrJw+jI2rRtTPNa8mkl4vav9sxnY=
github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM=
github.com/fxamacker/cbor/v2 v2.9.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ=
github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
@@ -313,8 +313,8 @@ github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 h1:n661drycOFuPLCN
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4=
github.com/shuffle/opensearch-go/v4 v4.0.0 h1:Mh85CD1MwOgXiFFYlzS1llnvdqL3CztRdR1ZT/SLIjU=
github.com/shuffle/opensearch-go/v4 v4.0.0/go.mod h1:gVLZKQE5khQWMb68XBtgKrhu78oLGL2zHwAGnFMDwC0=
github.com/shuffle/shuffle-shared v1.1.4 h1:TfeP9yslJfajdS2Sk/I81xU+R6ZxkyC49h9hc7oPBv4=
github.com/shuffle/shuffle-shared v1.1.4/go.mod h1:ycTfvlAIXv78Vxu+fHtPPIEP8mK9Pwtfk/yw2uMiAB0=
github.com/shuffle/shuffle-shared v1.2.24 h1:5jH7/QE4Lf+Yt55oPuszE93Ce34hn1lBVuyEdkX9nic=
github.com/shuffle/shuffle-shared v1.2.24/go.mod h1:RSKyexqkGDB+WbboGfWMj1Sfl4e49MI2CvW3pFmirg4=
github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
functions/onprem/orborus/orborus.go
+231 -56
View File
@@ -134,7 +134,83 @@ var window = shuffle.NewTimeWindow(1 * time.Minute)
func init() {
var err error
// Look for argc/argv and map environment variables
sensorMode := false
for _, arg := range os.Args {
if !strings.HasPrefix(arg, "--") {
continue
}
// Split away =
value := ""
if strings.Contains(arg, "=") {
newArg := strings.Split(arg, "=")[0]
value = strings.Split(arg, "=")[1]
arg = newArg
} else {
continue
}
if len(value) == 0 {
continue
}
parsedArg := strings.TrimPrefix(arg, "--")
parsedArg = strings.ReplaceAll(strings.ToUpper(parsedArg), " ", "_")
if !strings.HasPrefix(parsedArg, "SHUFFLE_") {
parsedArg = "SHUFFLE_" + parsedArg
}
if parsedArg == "SHUFFLE_SENSOR_MODE" {
parsedArg = "SHUFFLE_AGENT_SENSOR_MODE"
} else if parsedArg == "SHUFFLE_AGENT_MODE" {
parsedArg = "SHUFFLE_AGENT_SENSOR_MODE"
}
if parsedArg == "SHUFFLE_AGENT_SENSOR_MODE" && strings.ToLower(value) == "true" {
sensorMode = true
}
os.Setenv(parsedArg, value)
}
if sensorMode {
log.Printf("[INFO] Enabling sensormode (init check)")
for _, arg := range os.Args {
if !strings.HasPrefix(arg, "--") {
continue
}
// Split away =
value := ""
if strings.Contains(arg, "=") {
newArg := strings.Split(arg, "=")[0]
value = strings.Split(arg, "=")[1]
arg = newArg
} else {
continue
}
if len(value) == 0 {
continue
}
arg = strings.TrimPrefix(arg, "--")
if arg == "queue" {
os.Setenv("ENVIRONMENT_NAME", value)
environment = value
} else if arg == "auth" {
os.Setenv("AUTH", value)
auth = value
} else if arg == "org_id" {
os.Setenv("ORG", value)
org = value
}
}
} else {
// dockercli, err = dockerclient.NewEnvClient()
dockercli, dockerApiVersion, err = shuffle.GetDockerClient()
if err != nil {
@@ -173,7 +249,7 @@ func init() {
os.Setenv("SHUFFLE_PIPELINE_AUTH", pipelineApikey)
}
}
}
}
// form id of current running container
@@ -2014,7 +2090,15 @@ func parseResourceUsage(body io.Reader) (float64, float64, error) {
}
func getOrborusStats(ctx context.Context) shuffle.OrborusStats {
func getHostname() (string, error) {
hostname, err := os.Hostname()
if err != nil {
return "", fmt.Errorf("failed to get hostname: %w", err)
}
return hostname, nil
}
func getOrborusStats(ctx context.Context, sensorMode shuffle.SensorMode) shuffle.OrborusStats {
newStats := shuffle.OrborusStats{
OrgId: org,
Environment: environment,
@@ -2037,15 +2121,53 @@ func getOrborusStats(ctx context.Context) shuffle.OrborusStats {
return newStats
}
// FIXME: Returning for now due to this causing network congestion
// and database fillup. The backend api also has it disabled.
if sensorMode.Enabled {
newStats.SensorDetails.SensorMode = true
hostname, err := getHostname()
if err == nil {
newStats.SensorDetails.Hostname = hostname
}
newStats.SensorDetails.OS = runtime.GOOS
newStats.SensorDetails.Arch = runtime.GOARCH
newStats.SensorDetails.ElevatedAccess = shuffle.IsElevated()
newStats.SensorDetails.Serial = shuffle.GetProfiler()
if sensorMode.SoftwareListEnabled {
// Check cache first before running the command
newStats.SensorDetails.InstalledSoftware = shuffle.ListInstalledSoftware()
}
if sensorMode.HdEncryptedCheck {
newStats.SensorDetails.HdEncrypted = fmt.Sprintf("%t", shuffle.IsDiskEncrypted())
}
if sensorMode.ScreenlockCheck {
newStats.SensorDetails.AutomaticScreenlockEnabled = fmt.Sprintf("%t", shuffle.IsAutomaticScreenlockEnabled())
}
if len(sensorMode.LogForwarding) > 0 {
newStats.SensorDetails.LogForwarding = fmt.Sprintf("not implemented: %s", sensorMode.LogForwarding)
}
if sensorMode.ResponseActionsEnabled {
newStats.SensorDetails.ResponseActionsEnabled = fmt.Sprintf("not implemented: %s", sensorMode.ResponseActionsEnabled)
}
return newStats
} else {
return newStats
}
// FIXME: Should we reach here anymore? Can it be useful? Primarily used for stats.
// Disable orborus stats
if os.Getenv("SHUFFLE_STATS_DISABLED") == "true" {
return newStats
}
// FIXME: Returning for now due to this causing network congestion
// and database fillup. The backend api also has it disabled.
return newStats
// Use the docker API to get the CPU usage of the docker engine machine
pers, err := dockercli.Info(ctx)
if err != nil {
@@ -2240,17 +2362,30 @@ func cleanup() {
log.Printf("[INFO] Cleaning up during shutdown")
ctx := context.Background()
cleanupExistingNodes(ctx)
zombiecheck(ctx, 600)
zombiecheck(ctx, 600, shuffle.SensorMode{
Enabled: os.Getenv("SHUFFLE_AGENT_MODE") == "true",
})
os.Exit(0)
}
func StartAgent() {
log.Printf("[INFO] Starting Orborus agent mode")
func StartAgentSensor(sensorMode shuffle.SensorMode) error {
if sensorMode.Enabled == false {
return errors.New("Sensor mode is not enabled. Set SHUFFLE_AGENT_MODE to true to enable it.")
}
auditLogEnabled := os.Getenv("SHUFFLE_AUDIT_LOG_ENABLED") == "true"
log.Printf("[INFO] Starting Orborus - host monitoring mode (sensor/agent)")
if auditLogEnabled {
log.Printf("[INFO] Audit log monitoring is enabled")
// Check if inside Docker/Kubernetes. Use Docker/Kubernetes libraries
if isKubernetes == "true" || shuffle.IsRunningInCluster() {
log.Printf("[INFO] Detected Kubernetes environment. Not valid for Sensor Mode. Exiting.")
return errors.New("Kubernetes environment detected. Sensor mode is not valid in Kubernetes. Exiting.")
} else if swarmConfig == "run" || swarmConfig == "swarm" {
log.Printf("[INFO] Detected Docker Swarm environment. Not valid for Sensor Mode. Exiting.")
return errors.New("Docker Swarm environment detected. Sensor mode is not valid in Docker Swarm. Exiting.")
}
if len(sensorMode.LogForwarding) > 0 {
log.Printf("[INFO] Audit log monitoring is enabled (SHUFFLE_LOG_FORWARDING=<endpoint>)")
// Initialize telemetry configuration
telemetryConfig := shuffle.TelemetryConfig{
@@ -2299,26 +2434,59 @@ func StartAgent() {
}
}
} else {
log.Printf("[INFO] Audit log monitoring is disabled")
log.Printf("[INFO] Audit log monitoring is NOT enabled (SHUFFLE_LOG_FORWARDING=<endpoint>")
}
select {}
return nil
}
// Initial loop etc
func main() {
// Get arch. amd64 or arm64
//sigCh := make(chan os.Signal, 1)
//signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
//defer cleanup()
// Checks for whether sensor mode is enabled for detection/response
sensorMode := shuffle.SensorMode{
Enabled: os.Getenv("SHUFFLE_AGENT_SENSOR_MODE") == "true",
agentMode := os.Getenv("SHUFFLE_AGENT_MODE")
if agentMode == "true" {
log.Printf("[INFO] Running in agent mode. Starting the agent.")
StartAgent()
return
LogForwarding: os.Getenv("SHUFFLE_LOG_FORWARDING"),
SoftwareListEnabled: os.Getenv("SHUFFLE_SOFTWARE_LIST_ENABLED") == "true",
HdEncryptedCheck: os.Getenv("SHUFFLE_HD_ENCRYPTED_CHECK") == "true",
ScreenlockCheck: os.Getenv("SHUFFLE_SCREENLOCK_CHECK") == "true",
ResponseActionsEnabled: os.Getenv("SHUFFLE_RESPONSE_ACTIONS_ENABLED") == "true",
}
ctx := context.Background()
workerTimeout := 600
workerImage := fmt.Sprintf("ghcr.io/shuffle/shuffle-worker:%s", workerVersion)
if len(newWorkerImage) > 0 {
workerImage = newWorkerImage
}
if len(orborusUuid) == 0 {
orborusUuid = uuid.NewV4().String()
}
client := shuffle.GetExternalClient(baseUrl)
fullUrl := fmt.Sprintf("%s/api/v1/workflows/queue", baseUrl)
// Increases default concurrency to 50 for swarm
if maxConcurrency < 50 && (swarmConfig == "run" || swarmConfig == "swarm") {
fullUrl += "?amount=50"
}
if sensorMode.Enabled {
// Start high on purpose for now
if sleepTime < 30 {
sleepTime = 30
}
log.Printf("[INFO] Running in sensor/agent mode. Starting the agent.")
err := StartAgentSensor(sensorMode)
if err != nil {
log.Printf("[ERROR] Failed to start sensor/agent mode: %#v", err)
return
}
} else {
if os.Getenv("SHUFFLE_PIPELINE_STANDALONE") == "true" {
log.Printf("[INFO] Allowing use of standalone pipeline (tenzir). URL: %s", pipelineUrl)
@@ -2374,10 +2542,6 @@ func main() {
baseUrl = "https://shuffler.io"
}
if len(orborusUuid) == 0 {
orborusUuid = uuid.NewV4().String()
}
//if orgId == "" {
// log.Printf("[ERROR] Org not defined. Set variable ORG_ID based on your org")
// os.Exit(3)
@@ -2416,7 +2580,6 @@ func main() {
os.Setenv("SHUFFLE_APP_SDK_TIMEOUT", "120")
}
workerTimeout := 600
if workerTimeoutEnv != "" {
tmpInt, err := strconv.Atoi(workerTimeoutEnv)
if err == nil {
@@ -2446,7 +2609,6 @@ func main() {
log.Printf(`[DEBUG] Running docker with default socket /var/run/docker.sock or `)
}
ctx := context.Background()
// Run by default from now
//commenting for now as its stoppoing minikube
@@ -2485,10 +2647,6 @@ func main() {
log.Printf("[INFO] Setting up Docker environment. Downloading worker and App SDK!")
initializeImages()
workerImage := fmt.Sprintf("ghcr.io/shuffle/shuffle-worker:%s", workerVersion)
if len(newWorkerImage) > 0 {
workerImage = newWorkerImage
}
if swarmConfig == "run" || swarmConfig == "swarm" || isKubernetes == "true" {
@@ -2523,21 +2681,14 @@ func main() {
//deployServiceWorkers(workerImage)
}
zombiecheck(ctx, workerTimeout)
client := shuffle.GetExternalClient(baseUrl)
fullUrl := fmt.Sprintf("%s/api/v1/workflows/queue", baseUrl)
// Increases default concurrency to 50 for swarm
if maxConcurrency < 50 && (swarmConfig == "run" || swarmConfig == "swarm") {
fullUrl += "?amount=50"
}
zombiecheck(ctx, workerTimeout, sensorMode)
if isKubernetes == "true" {
log.Printf("[INFO] Finished configuring kubernetes environment. Connecting to %s", fullUrl)
} else {
log.Printf("[INFO] Finished configuring docker environment. Connecting to %s", fullUrl)
}
}
forwardData := bytes.NewBuffer([]byte{})
forwardMethod := "POST"
@@ -2591,21 +2742,21 @@ func main() {
swarmControlMode = true
}
log.Printf("[INFO] Waiting for executions at %s with Environment %#v", fullUrl, environment)
log.Printf("[INFO] Waiting for executions at %s with Environment %#v. Sensormode: %#v", fullUrl, environment, sensorMode.Enabled)
hasStarted := false
for {
if req.Method == "POST" {
if req.Method == "POST" && !sensorMode.Enabled {
// Should find data to send (memory etc.)
// Create timeout of max 4 seconds just in case
// Create timeout of max a few seconds just in case
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
// Marshal and set body
orborusStats := getOrborusStats(ctx)
orborusStats := getOrborusStats(ctx, sensorMode)
pipelinePayload, pipelineerr := sendPipelineHealthStatus()
pipelinePayload, pipelineerr := sendPipelineHealthStatus(sensorMode)
if pipelineerr != nil {
// Too verbose to be enabled.
@@ -2625,6 +2776,17 @@ func main() {
time.Sleep(time.Duration(sleepTime) * time.Second)
continue
}
} else if sensorMode.Enabled {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
orborusStats := getOrborusStats(ctx, sensorMode)
jsonData, err := json.Marshal(orborusStats)
if err == nil {
req.Body = ioutil.NopCloser(bytes.NewBuffer(jsonData))
} else {
log.Printf("[ERROR] Failed marshalling. Maybe max 4 second timeout? %s", err)
}
}
newresp, err := client.Do(req)
@@ -2633,7 +2795,7 @@ func main() {
zombiecounter += 1
if zombiecounter*sleepTime > workerTimeout {
go zombiecheck(ctx, workerTimeout)
go zombiecheck(ctx, workerTimeout, sensorMode)
zombiecounter = 0
}
time.Sleep(time.Duration(sleepTime) * time.Second)
@@ -2656,7 +2818,7 @@ func main() {
log.Printf("[ERROR] Failed reading body from Shuffle: %s", err)
zombiecounter += 1
if zombiecounter*sleepTime > workerTimeout {
go zombiecheck(ctx, workerTimeout)
go zombiecheck(ctx, workerTimeout, sensorMode)
zombiecounter = 0
}
time.Sleep(time.Duration(sleepTime) * time.Second)
@@ -2672,7 +2834,7 @@ func main() {
log.Printf("[ERROR] Backend connection failed for url '%s', or is missing (%d): %s", fullUrl, newresp.StatusCode, string(body))
} else {
if !hasStarted {
log.Printf("[DEBUG] Starting iteration on environment %#v (default = Shuffle). Got statuscode %d from backend on first request", environment, newresp.StatusCode)
log.Printf("[DEBUG] Starting iteration on environment %#v (default: Shuffle). Got statuscode %d from backend on first request", environment, newresp.StatusCode)
}
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" && os.Getenv("SHUFFLE_SCALE_REPLICAS") == "" {
@@ -2688,7 +2850,7 @@ func main() {
sleepTime = 10
zombiecounter += 1
if zombiecounter*sleepTime > workerTimeout {
go zombiecheck(ctx, workerTimeout)
go zombiecheck(ctx, workerTimeout, sensorMode)
zombiecounter = 0
}
time.Sleep(time.Duration(sleepTime) * time.Second)
@@ -2700,7 +2862,6 @@ func main() {
// Type string `json:"type"`
}
// FIXME: Add features here for orborus & worker to
// do things on behalf of backend
var toBeRemoved shuffle.ExecutionRequestWrapper
if len(executionRequests.Data) > 0 {
@@ -2733,7 +2894,12 @@ func main() {
executionRequests.Data = deduplicatedJobs
for _, incRequest := range executionRequests.Data {
if sensorMode.Enabled {
log.Printf("[DEBUG] Sensor mode enabled. Removing job from queue without processing: %#v", incRequest)
toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
} else {
// Looking for specific jobs
if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_START" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" || incRequest.Type == "PIPELINE_UPDATE" {
log.Printf("[INFO] Handling pipeline request from backend: '%s' with argument '%s'", incRequest.Type, incRequest.ExecutionArgument)
@@ -2846,6 +3012,7 @@ func main() {
newrequests = append(newrequests, incRequest)
}
}
}
if len(toBeRemoved.Data) > 0 {
err = sendRemoveRequest(client, toBeRemoved, baseUrl, environment, auth, org, sleepTime)
@@ -2865,7 +3032,7 @@ func main() {
if len(executionRequests.Data) == 0 {
zombiecounter += 1
if zombiecounter*sleepTime > workerTimeout {
go zombiecheck(ctx, workerTimeout)
go zombiecheck(ctx, workerTimeout, sensorMode)
zombiecounter = 0
}
time.Sleep(time.Duration(sleepTime) * time.Second)
@@ -2876,7 +3043,7 @@ func main() {
executionCount = getRunningWorkers(ctx, workerTimeout)
if executionCount >= maxConcurrency {
if zombiecounter*sleepTime > workerTimeout {
go zombiecheck(ctx, workerTimeout)
go zombiecheck(ctx, workerTimeout, sensorMode)
zombiecounter = 0
}
time.Sleep(time.Duration(sleepTime) * time.Second)
@@ -4019,7 +4186,11 @@ func removePath(containerName, path string) error {
return nil
}
func sendPipelineHealthStatus() (shuffle.LakeConfig, error) {
func sendPipelineHealthStatus(sensorMode shuffle.SensorMode) (shuffle.LakeConfig, error) {
if sensorMode.Enabled {
return shuffle.LakeConfig{}, nil
}
pipelinePayload := shuffle.LakeConfig{
Enabled: false,
Pipelines: []shuffle.PipelineInfo{},
@@ -4218,7 +4389,11 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int {
// FIXME - add this to remove exited workers
// Should it check what happened to the execution? idk
func zombiecheck(ctx context.Context, workerTimeout int) error {
func zombiecheck(ctx context.Context, workerTimeout int, sensorMode shuffle.SensorMode) error {
if sensorMode.Enabled {
return nil
}
isK8s := isKubernetes == "true"
executionIds = []string{}