feat: add proxy job

.proto/redcarbon/agents_public/v1/types.proto

@@ -2,6 +2,8 @@ syntax = "proto3";
 
 package redcarbon.agents_public.v1;
 
+import "google/protobuf/duration.proto";
+
 message AgentConfiguration {
   optional QRadarJobConfiguration qradar_job_configuration = 1;
   optional SentinelOneJobConfiguration sentinelone_job_configuration = 2;
@@ -26,3 +28,22 @@ message FortiSIEMJobConfiguration {
   string password = 3;
   bool verify_ssl = 4;
 }
+
+message ValueHeader {
+  repeated string values = 1;
+}
+
+message AgentRequest {
+  string request_id = 1;
+  string method = 2;
+  string url = 3;
+  map<string, ValueHeader> headers = 4;
+  bytes body = 5;
+  google.protobuf.Duration timeout = 6;
+}
+
+message AgentResponse {
+  int32 status = 1;
+  map<string, ValueHeader> headers = 2;
+  bytes body = 3;
+}

.proto/redcarbon/agents_public/v1/v1.proto

@@ -9,6 +9,8 @@ service AgentsPublicAPIsV1Srv {
   rpc HZ(HZRequest) returns (HZResponse) {}
   rpc IngestIncident(IngestIncidentRequest) returns (IngestIncidentResponse) {}
   rpc FetchAgentConfiguration(FetchAgentConfigurationRequest) returns (FetchAgentConfigurationResponse) {}
+  rpc FetchAgentRequests(FetchAgentRequestsRequest) returns (FetchAgentRequestsResponse) {}
+  rpc SubmitAgentResponse(SubmitAgentResponseRequest) returns (SubmitAgentResponseResponse) {}
 }
 
 message HZRequest {
@@ -37,3 +39,16 @@ message FetchAgentConfigurationRequest {}
 message FetchAgentConfigurationResponse {
   AgentConfiguration configuration = 1;
 }
+
+message FetchAgentRequestsRequest {}
+
+message FetchAgentRequestsResponse {
+  repeated AgentRequest requests = 1;
+}
+
+message SubmitAgentResponseRequest {
+  string request_id = 1;
+  AgentResponse response = 2;
+}
+
+message SubmitAgentResponseResponse {}

cmd/main.go

@@ -3,17 +3,18 @@ package main
 import (
 	"context"
 	"errors"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
 	"github.com/go-co-op/gocron"
 	"github.com/google/go-github/v50/github"
 	"golang.org/x/sync/errgroup"
-	"os"
-	"os/signal"
 	"pkg.redcarbon.ai/cmd/profile"
 	"pkg.redcarbon.ai/internal/cli"
 	"pkg.redcarbon.ai/internal/config"
 	"pkg.redcarbon.ai/internal/routines"
-	"syscall"
-	"time"
 
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -27,6 +28,7 @@ const (
 	configRoutineInterval = "10m"
 	updateRoutineInterval = "1d"
 	debugRoutineInterval  = "2s"
+	proxyRoutineInterval  = "500ms"
 
 	updateErrorCode = 3
 )
@@ -79,9 +81,8 @@ func run(cmd *cobra.Command, args []string) {
 		logrus.Fatal("No profiles found, please add one by running `redcarbon profile add`")
 	}
 
-	ctx, cancelFn := signal.NotifyContext(context.Background(), syscall.SIGINT)
-
-	defer cancelFn()
+	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT)
+	defer cancel()
 
 	g, ctx := errgroup.WithContext(ctx)
 
@@ -101,6 +102,7 @@ func run(cmd *cobra.Command, args []string) {
 			s.Every(updateRoutineInterval).StartImmediately().SingletonMode().Do(r.UpdateRoutine, ctx)
 			s.Every(hzRoutineInterval).StartImmediately().Do(r.HZRoutine, ctx)
 			s.Every(configRoutine).StartImmediately().SingletonMode().Do(r.ConfigRoutine, ctx)
+			s.Every(proxyRoutineInterval).StartImmediately().Do(r.ProxyRoutine, ctx)
 
 			return nil
 		})

internal/routines/proxy.go

@@ -0,0 +1,184 @@
+package routines
+
+import (
+	"bytes"
+	"context"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"connectrpc.com/connect"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/protobuf/types/known/durationpb"
+	agents_publicv1 "pkg.redcarbon.ai/proto/redcarbon/agents_public/v1"
+)
+
+var defaultTimeout = 1 * time.Minute
+
+var httpCli = &http.Client{
+	Timeout: defaultTimeout,
+	Transport: &http.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+	},
+}
+
+func (r RoutineConfig) ProxyRoutine(ctx context.Context) {
+	logrus.Debug("Starting the proxy routine...")
+
+	// Prepare the request to fetch agent requests
+	req := connect.NewRequest(&agents_publicv1.FetchAgentRequestsRequest{})
+	req.Header().Set("authorization", fmt.Sprintf("ApiToken %s", r.profile.Profile.Token))
+
+	ctx, cancel := context.WithTimeout(ctx, defaultTimeout)
+	defer cancel()
+
+	logrus.Debug("Fetching the agent requests...")
+	res, err := r.agentsCli.FetchAgentRequests(ctx, req)
+	if err != nil {
+		logrus.WithError(err).Error("Error while fetching the agent requests")
+		return
+	}
+
+	logrus.Debugf("Running %d agent requests...", len(res.Msg.Requests))
+
+	var wg sync.WaitGroup
+
+	for _, agentReq := range res.Msg.Requests {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			r.processRequest(ctx, agentReq)
+		}()
+	}
+
+	wg.Wait()
+
+	logrus.Debug("Proxy routine completed")
+}
+
+func (r RoutineConfig) processRequest(ctx context.Context, req *agents_publicv1.AgentRequest) {
+	l := logrus.WithFields(logrus.Fields{
+		"id": req.RequestId,
+	})
+
+	l.Debug("Handling request...")
+
+	ctx, cancel := extractTimeout(ctx, req.Timeout)
+	defer cancel()
+
+	httpReq, err := r.createHTTPProxyRequest(ctx, req)
+	if err != nil {
+		l.WithError(err).Error("Error while creating the HTTP request")
+		r.sendErrorToServer(ctx, req, "error while creating the HTTP request")
+		return
+	}
+
+	logrus.WithField("headers", httpReq.Header).Infof("Executing HTTP request: %s %s", req.Method, req.Url)
+
+	httpRes, err := httpCli.Do(httpReq)
+	if err != nil {
+		l.WithError(err).Error("Error while executing the HTTP request")
+		r.sendErrorToServer(ctx, req, "error while executing the HTTP request")
+		return
+	}
+	defer httpRes.Body.Close()
+
+	l.Infof("Request completed with status code %d, sending the response to the server", httpRes.StatusCode)
+
+	err = r.sendResponseToServer(ctx, req, httpRes)
+	if err != nil {
+		l.WithError(err).Error("Error while sending the response to the server")
+		r.sendErrorToServer(ctx, req, "error while sending the response to the server")
+		return
+	}
+}
+
+func extractTimeout(ctx context.Context, requestTimeout *durationpb.Duration) (context.Context, context.CancelFunc) {
+	timeout := 10 * time.Second
+	if requestTimeout != nil && requestTimeout.AsDuration() > 0 {
+		timeout = requestTimeout.AsDuration()
+	}
+
+	return context.WithTimeout(ctx, timeout)
+}
+
+func (r RoutineConfig) createHTTPProxyRequest(ctx context.Context, req *agents_publicv1.AgentRequest) (*http.Request, error) {
+	// Url is already validated by the server
+	httpReq, err := http.NewRequestWithContext(ctx, req.Method, req.Url, bytes.NewBuffer(req.Body))
+	if err != nil {
+		return nil, err
+	}
+
+	for key, value := range req.Headers {
+		if strings.ToLower(key) == "accept-encoding" {
+			continue
+		}
+
+		for _, v := range value.Values {
+			httpReq.Header.Add(key, v)
+		}
+	}
+
+	return httpReq, nil
+}
+
+func (r RoutineConfig) sendErrorToServer(ctx context.Context, req *agents_publicv1.AgentRequest, reason string) {
+	body, err := json.Marshal(map[string]string{
+		"error": reason,
+	})
+	if err != nil {
+		logrus.WithError(err).Error("Error while marshalling the error to the server")
+		return
+	}
+
+	response := connect.NewRequest(&agents_publicv1.SubmitAgentResponseRequest{
+		RequestId: req.RequestId,
+		Response: &agents_publicv1.AgentResponse{
+			Status:  int32(http.StatusInternalServerError),
+			Body:    body,
+			Headers: map[string]*agents_publicv1.ValueHeader{},
+		},
+	})
+	response.Header().Set("authorization", fmt.Sprintf("ApiToken %s", r.profile.Profile.Token))
+
+	_, err = r.agentsCli.SubmitAgentResponse(ctx, response)
+	if err != nil {
+		logrus.WithError(err).Error("Error while sending the error to the server")
+	}
+}
+
+func (r RoutineConfig) sendResponseToServer(ctx context.Context, req *agents_publicv1.AgentRequest, httpRes *http.Response) error {
+	body, err := io.ReadAll(httpRes.Body)
+	if err != nil {
+		return err
+	}
+
+	headers := make(map[string]*agents_publicv1.ValueHeader)
+	for key, values := range httpRes.Header {
+		headers[key] = &agents_publicv1.ValueHeader{
+			Values: values,
+		}
+	}
+
+	response := connect.NewRequest(&agents_publicv1.SubmitAgentResponseRequest{
+		RequestId: req.RequestId,
+		Response: &agents_publicv1.AgentResponse{
+			Status:  int32(httpRes.StatusCode),
+			Body:    body,
+			Headers: headers,
+		},
+	})
+	response.Header().Set("authorization", fmt.Sprintf("ApiToken %s", r.profile.Profile.Token))
+
+	_, err = r.agentsCli.SubmitAgentResponse(ctx, response)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

mise.toml

@@ -0,0 +1,3 @@
+[tools]
+buf = "1.45.0"
+go = "1.24"