diff --git a/backend/backend/apps/api-gateway/conduit.go b/backend/backend/apps/api-gateway/conduit.go new file mode 100644 index 0000000..035b096 --- /dev/null +++ b/backend/backend/apps/api-gateway/conduit.go @@ -0,0 +1,51 @@ +package main + +import ( + "context" + "net/http" + "os" + "time" + + conduit "github.com/devXyi/prexus-intelligence/backend/integrations/conduit" + "github.com/gin-gonic/gin" +) + +var prexusConduit *conduit.Client + +func initConduit() { + if os.Getenv("CONDUIT_CLIENT_ID") == "" || os.Getenv("CONDUIT_CLIENT_SECRET") == "" { + return + } + cfg, err := conduit.LoadConfig() + if err != nil { + return + } + prexusConduit = conduit.NewClient(cfg) +} + +// handleConduitTools is intentionally read-only. It proves the external +// Conduit integration without moving any Prexus risk traffic onto it yet. +func handleConduitTools(c *gin.Context) { + if prexusConduit == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "Conduit integration is not configured"}) + return + } + ctx, cancel := context.WithTimeout(c.Request.Context(), 20*time.Second) + defer cancel() + + if _, err := prexusConduit.Initialize(ctx); err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": "Conduit initialization failed"}) + return + } + if err := prexusConduit.Initialized(ctx); err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": "Conduit session initialization failed"}) + return + } + tools, err := prexusConduit.ListTools(ctx) + if err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": "Conduit tools/list failed"}) + return + } + + c.Data(http.StatusOK, "application/json", tools) +} diff --git a/backend/backend/apps/api-gateway/main.go b/backend/backend/apps/api-gateway/main.go index 3e32ff3..36351e4 100644 --- a/backend/backend/apps/api-gateway/main.go +++ b/backend/backend/apps/api-gateway/main.go @@ -25,17 +25,12 @@ import ( const VERSION = "2.1.0" var allowedModels = map[string]struct{}{ - "claude-opus-4-7": {}, - "claude-sonnet-4-6": {}, - "claude-haiku-4-5-20251001": {}, + "claude-opus-4-7": {}, "claude-sonnet-4-6": {}, "claude-haiku-4-5-20251001": {}, } const defaultModel = "claude-haiku-4-5-20251001" -type ipLimiter struct { - limiter *rate.Limiter - lastSeen time.Time -} +type ipLimiter struct { limiter *rate.Limiter; lastSeen time.Time } var ( limiters = make(map[string]*ipLimiter) @@ -45,10 +40,7 @@ var ( func getLimiter(ip string) *rate.Limiter { limitersMu.Lock() defer limitersMu.Unlock() - if il, ok := limiters[ip]; ok { - il.lastSeen = time.Now() - return il.limiter - } + if il, ok := limiters[ip]; ok { il.lastSeen = time.Now(); return il.limiter } l := rate.NewLimiter(5, 10) limiters[ip] = &ipLimiter{limiter: l, lastSeen: time.Now()} return l @@ -62,9 +54,7 @@ func cleanupLimiters(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: limitersMu.Lock() - for ip, il := range limiters { - if time.Since(il.lastSeen) > 10*time.Minute { delete(limiters, ip) } - } + for ip, il := range limiters { if time.Since(il.lastSeen) > 10*time.Minute { delete(limiters, ip) } } limitersMu.Unlock() } } @@ -72,8 +62,7 @@ func cleanupLimiters(ctx context.Context) { func RateLimitMiddleware() gin.HandlerFunc { return func(c *gin.Context) { - ip := c.ClientIP() - if !getLimiter(ip).Allow() { + if !getLimiter(c.ClientIP()).Allow() { c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{"error": "rate limit exceeded — slow down"}) return } @@ -84,18 +73,13 @@ func RateLimitMiddleware() gin.HandlerFunc { const maxBodyBytes = 1 << 20 func BodySizeMiddleware() gin.HandlerFunc { - return func(c *gin.Context) { - c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, maxBodyBytes) - c.Next() - } + return func(c *gin.Context) { c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, maxBodyBytes); c.Next() } } func main() { _ = godotenv.Load() - port := os.Getenv("PORT") - if port == "" { port = "8080" } - env := os.Getenv("ENV") - if env == "production" { gin.SetMode(gin.ReleaseMode) } + port := os.Getenv("PORT"); if port == "" { port = "8080" } + env := os.Getenv("ENV"); if env == "production" { gin.SetMode(gin.ReleaseMode) } dataEngineURL := getDataEngineURL() if dataEngineURL == "" { log.Fatal("DATA_ENGINE_URL is not set — cannot start") } @@ -103,6 +87,8 @@ func main() { defer CloseDB() log.Printf("✓ Database connected") log.Printf("✓ Data engine: %s", dataEngineURL) + initConduit() + if prexusConduit != nil { log.Printf("✓ Conduit MCP integration configured") } else { log.Printf("• Conduit MCP integration not configured (optional)") } allowedOrigins := getAllowedOrigins() log.Printf("✓ CORS origins: %v", allowedOrigins) @@ -110,19 +96,11 @@ func main() { defer cancel() go cleanupLimiters(ctx) - r := gin.New() - r.Use(gin.Recovery()) - r.Use(RequestID()) - r.Use(BodySizeMiddleware()) - r.Use(requestLogger()) - + r := gin.New(); r.Use(gin.Recovery()); r.Use(RequestID()); r.Use(BodySizeMiddleware()); r.Use(requestLogger()) if tp := os.Getenv("TRUSTED_PROXIES"); tp != "" { if err := r.SetTrustedProxies(strings.Split(tp, ",")); err != nil { log.Fatalf("Invalid TRUSTED_PROXIES: %v", err) } log.Printf("✓ Trusted proxies: %s", tp) - } else { - _ = r.SetTrustedProxies(nil) - log.Printf("✓ Trusted proxies: none (direct connections only)") - } + } else { _ = r.SetTrustedProxies(nil); log.Printf("✓ Trusted proxies: none (direct connections only)") } r.Use(cors.New(cors.Config{ AllowOrigins: allowedOrigins, @@ -133,16 +111,7 @@ func main() { MaxAge: 12 * time.Hour, })) - // ── Public Routes ───────────────────────────────────── - r.GET("/", func(c *gin.Context) { - c.JSON(http.StatusOK, gin.H{ - "service": "prexus-api-gateway", - "status": "ok", - "version": VERSION, - "health": "/health", - "docs": "API gateway endpoints are documented by the service contract", - }) - }) + r.GET("/", func(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"service": "prexus-api-gateway", "status": "ok", "version": VERSION, "health": "/health", "docs": "API gateway endpoints are documented by the service contract"}) }) r.GET("/health", RateLimitMiddleware(), handleHealth) r.POST("/register", RateLimitMiddleware(), handleRegister) r.POST("/login", RateLimitMiddleware(), handleLogin) @@ -158,28 +127,24 @@ func main() { auth.POST("/risk/portfolio", RequirePermission("risk:run"), proxyToDataEngine("/risk/portfolio")) auth.POST("/risk/stress-test", RequirePermission("risk:run"), proxyToDataEngine("/risk/stress-test")) auth.GET("/risk/health", RequirePermission("risk:run"), proxyToDataEngineGET("/risk/health")) - auth.GET("/sources", RequirePermission("risk:run"), proxyToDataEngineGET("/sources")) auth.GET("/lake/stats", RequirePermission("risk:run"), proxyToDataEngineGET("/lake/stats")) auth.GET("/lake/files", RequirePermission("risk:run"), proxyToDataEngineGET("/lake/files")) - auth.POST("/chat", RateLimitMiddleware(), RequirePermission("risk:run"), proxyToDataEngine("/chat")) auth.POST("/claude", RateLimitMiddleware(), RequirePermission("risk:run"), handleClaude) auth.POST("/analyze", RateLimitMiddleware(), RequirePermission("risk:run"), proxyToDataEngine("/analyze")) + auth.GET("/conduit/tools", RequirePermission("conduit:read"), handleConduitTools) auth.GET("/me", handleGetMe) auth.PUT("/me", handleUpdateMe) } log.Printf("🚀 Prexus API Gateway v%s running on :%s (env=%s)", VERSION, port, env) srv := &http.Server{Addr: ":" + port, Handler: r} - go func() { - if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Fatalf("Server error: %v", err) } - }() + go func() { if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Fatalf("Server error: %v", err) } }() <-ctx.Done() log.Println("Shutting down gracefully…") - shutCtx, shutCancel := context.WithTimeout(context.Background(), 15*time.Second) - defer shutCancel() + shutCtx, shutCancel := context.WithTimeout(context.Background(), 15*time.Second); defer shutCancel() if err := srv.Shutdown(shutCtx); err != nil { log.Printf("Graceful shutdown error: %v", err) } log.Println("Server stopped.") } @@ -188,34 +153,22 @@ func handleClaude(c *gin.Context) { c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, maxBodyBytes) var req struct { Message string `json:"message" binding:"required"`; Model string `json:"model"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}); return } - model := strings.TrimSpace(req.Model) - if model == "" { model = defaultModel } - if _, ok := allowedModels[model]; !ok { - c.JSON(http.StatusBadRequest, gin.H{"error": "unsupported model", "allowed": getAllowedModelList()}) - return - } - userID, _ := c.Get("user_id") - reqID, _ := c.Get("request_id") + model := strings.TrimSpace(req.Model); if model == "" { model = defaultModel } + if _, ok := allowedModels[model]; !ok { c.JSON(http.StatusBadRequest, gin.H{"error": "unsupported model", "allowed": getAllowedModelList()}); return } + userID, _ := c.Get("user_id"); reqID, _ := c.Get("request_id") log.Printf("[claude] req=%v user=%v model=%s ip=%s msg_len=%d", reqID, userID, model, c.ClientIP(), len(req.Message)) - ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) - defer cancel() + ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second); defer cancel() reply, err := AnalyzeProbability(ctx, req.Message, model) if err != nil { log.Printf("[claude] error req=%v user=%v: %v", reqID, userID, err); c.JSON(http.StatusInternalServerError, gin.H{"error": "AI inference failed"}); return } c.JSON(http.StatusOK, gin.H{"reply": reply}) } -func handleHealth(c *gin.Context) { - c.JSON(http.StatusOK, gin.H{"status": "ok", "service": "prexus-api-gateway", "version": VERSION, "timestamp": time.Now().UTC().Format(time.RFC3339)}) -} +func handleHealth(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"status": "ok", "service": "prexus-api-gateway", "version": VERSION, "timestamp": time.Now().UTC().Format(time.RFC3339)}) } func requestLogger() gin.HandlerFunc { return func(c *gin.Context) { - start := time.Now() - c.Next() - latency := time.Since(start) - userID, exists := c.Get("user_id") - if !exists { userID = "anonymous" } - reqID, _ := c.Get("request_id") + start := time.Now(); c.Next(); latency := time.Since(start) + userID, exists := c.Get("user_id"); if !exists { userID = "anonymous" }; reqID, _ := c.Get("request_id") log.Printf("[%d] %s %s req=%v user=%v ip=%s latency=%v", c.Writer.Status(), c.Request.Method, c.Request.URL.Path, reqID, userID, c.ClientIP(), latency) } } @@ -223,26 +176,12 @@ func requestLogger() gin.HandlerFunc { func getAllowedOrigins() []string { raw := os.Getenv("ALLOWED_ORIGINS") if raw == "" { log.Println("⚠️ ALLOWED_ORIGINS not set — defaulting to localhost (dev only)"); return []string{"http://localhost:3000", "http://localhost:5173"} } - origins := []string{} - for _, o := range strings.Split(raw, ",") { o = strings.TrimSpace(o); if o != "" { origins = append(origins, o) } } - return origins + origins := []string{}; for _, o := range strings.Split(raw, ",") { o = strings.TrimSpace(o); if o != "" { origins = append(origins, o) } }; return origins } -func getAllowedModelList() []string { - list := make([]string, 0, len(allowedModels)) - for m := range allowedModels { list = append(list, m) } - sort.Strings(list) - return list -} +func getAllowedModelList() []string { list := make([]string, 0, len(allowedModels)); for m := range allowedModels { list = append(list, m) }; sort.Strings(list); return list } -func RequestID() gin.HandlerFunc { - return func(c *gin.Context) { - id := fmt.Sprintf("%d", time.Now().UnixNano()) - c.Set("request_id", id) - c.Writer.Header().Set("X-Request-ID", id) - c.Next() - } -} +func RequestID() gin.HandlerFunc { return func(c *gin.Context) { id := fmt.Sprintf("%d", time.Now().UnixNano()); c.Set("request_id", id); c.Writer.Header().Set("X-Request-ID", id); c.Next() } } func init() { if os.Getenv("ENV") == "production" { return } diff --git a/backend/backend/apps/api-gateway/rbac.go b/backend/backend/apps/api-gateway/rbac.go index d6f3f23..3b12c33 100644 --- a/backend/backend/apps/api-gateway/rbac.go +++ b/backend/backend/apps/api-gateway/rbac.go @@ -5,53 +5,33 @@ package main import "github.com/gin-gonic/gin" -// ───────────────────────────────────────────────────────────── -// Role → Permissions Mapping -// ───────────────────────────────────────────────────────────── - var rolePermissions = map[string][]string{ "admin": { - "assets:read", - "assets:create", - "assets:update", - "assets:delete", - "risk:run", - "user:read", + "assets:read", "assets:create", "assets:update", "assets:delete", + "risk:run", "user:read", "conduit:read", }, - "user": { - "assets:read", - "assets:create", - "assets:update", - "risk:run", + "assets:read", "assets:create", "assets:update", "risk:run", "conduit:read", }, - "viewer": { "assets:read", }, } -// ───────────────────────────────────────────────────────────── -// Permission Check Middleware -// ───────────────────────────────────────────────────────────── - func RequirePermission(permission string) gin.HandlerFunc { return func(c *gin.Context) { role := c.GetString("role") - perms, ok := rolePermissions[role] if !ok { c.AbortWithStatusJSON(403, gin.H{"error": "Invalid role"}) return } - for _, p := range perms { if p == permission { c.Next() return } } - c.AbortWithStatusJSON(403, gin.H{"error": "Permission denied"}) } -} \ No newline at end of file +} diff --git a/backend/backend/integrations/conduit/client.go b/backend/backend/integrations/conduit/client.go new file mode 100644 index 0000000..0cad4fb --- /dev/null +++ b/backend/backend/integrations/conduit/client.go @@ -0,0 +1,205 @@ +package conduit + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" +) + +type Client struct { + cfg Config + httpClient *http.Client + token string + tokenExp time.Time + sessionID string + mu sync.Mutex + seq uint64 +} + +func NewClient(cfg Config) *Client { + return &Client{cfg: cfg, httpClient: &http.Client{Timeout: 30 * time.Second}} +} + +type tokenResponse struct { + AccessToken string `json:"access_token"` + TokenType string `json:"token_type"` + ExpiresIn int `json:"expires_in"` +} + +func (c *Client) accessToken(ctx context.Context) (string, error) { + c.mu.Lock() + if c.token != "" && time.Now().Before(c.tokenExp.Add(-30*time.Second)) { + t := c.token + c.mu.Unlock() + return t, nil + } + c.mu.Unlock() + + payload, err := json.Marshal(map[string]string{ + "client_id": c.cfg.ClientID, "client_secret": c.cfg.ClientSecret, + "audience": c.cfg.Audience, "grant_type": "client_credentials", + }) + if err != nil { + return "", fmt.Errorf("encode conduit token request: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.cfg.TokenURL, bytes.NewReader(payload)) + if err != nil { + return "", fmt.Errorf("create conduit token request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + resp, err := c.httpClient.Do(req) + if err != nil { + return "", fmt.Errorf("request conduit token: %w", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return "", fmt.Errorf("conduit token endpoint returned %s: %s", resp.Status, strings.TrimSpace(string(body))) + } + var tr tokenResponse + if err := json.Unmarshal(body, &tr); err != nil { + return "", fmt.Errorf("decode conduit token response: %w", err) + } + if tr.AccessToken == "" { + return "", fmt.Errorf("conduit token endpoint returned no access_token") + } + expires := time.Duration(tr.ExpiresIn) * time.Second + if expires <= 0 { + expires = 5 * time.Minute + } + c.mu.Lock() + c.token, c.tokenExp = tr.AccessToken, time.Now().Add(expires) + t := c.token + c.mu.Unlock() + return t, nil +} + +type rpcRequest struct { + JSONRPC string `json:"jsonrpc"` + ID uint64 `json:"id,omitempty"` + Method string `json:"method"` + Params interface{} `json:"params,omitempty"` +} + +type rpcResponse struct { + JSONRPC string `json:"jsonrpc"` + ID uint64 `json:"id"` + Result json.RawMessage `json:"result,omitempty"` + Error *rpcError `json:"error,omitempty"` +} + +type rpcError struct { + Code int `json:"code"` + Message string `json:"message"` + Data any `json:"data,omitempty"` +} + +func (c *Client) nextID() uint64 { return atomic.AddUint64(&c.seq, 1) } + +func (c *Client) request(ctx context.Context, method string, params interface{}, notification bool) (json.RawMessage, error) { + token, err := c.accessToken(ctx) + if err != nil { + return nil, err + } + rpc := rpcRequest{JSONRPC: "2.0", Method: method, Params: params} + if !notification { + rpc.ID = c.nextID() + } + body, err := json.Marshal(rpc) + if err != nil { + return nil, fmt.Errorf("encode MCP request: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.cfg.MCPURL, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("create MCP request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + c.mu.Lock() + if c.sessionID != "" { + req.Header.Set("Mcp-Session-Id", c.sessionID) + } + c.mu.Unlock() + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("MCP request failed: %w", err) + } + defer resp.Body.Close() + if sid := resp.Header.Get("Mcp-Session-Id"); sid != "" { + c.mu.Lock() + c.sessionID = sid + c.mu.Unlock() + } + responseBody, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("MCP endpoint returned %s: %s", resp.Status, strings.TrimSpace(string(responseBody))) + } + if notification { + return nil, nil + } + result, err := decodeRPC(responseBody, resp.Header.Get("Content-Type")) + if err != nil { + return nil, err + } + if result.Error != nil { + return nil, fmt.Errorf("MCP error %d: %s", result.Error.Code, result.Error.Message) + } + return result.Result, nil +} + +func (c *Client) call(ctx context.Context, method string, params interface{}) (json.RawMessage, error) { + return c.request(ctx, method, params, false) +} + +func (c *Client) notify(ctx context.Context, method string, params interface{}) error { + _, err := c.request(ctx, method, params, true) + return err +} + +func decodeRPC(body []byte, contentType string) (rpcResponse, error) { + var direct rpcResponse + if json.Unmarshal(body, &direct) == nil && (direct.Result != nil || direct.Error != nil) { + return direct, nil + } + for _, line := range strings.Split(string(body), "\n") { + line = strings.TrimSpace(line) + if !strings.HasPrefix(line, "data:") { + continue + } + payload := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + var msg rpcResponse + if json.Unmarshal([]byte(payload), &msg) == nil && (msg.Result != nil || msg.Error != nil) { + return msg, nil + } + } + return rpcResponse{}, fmt.Errorf("unable to decode MCP JSON-RPC response (content-type %q)", contentType) +} + +func (c *Client) Initialize(ctx context.Context) (json.RawMessage, error) { + return c.call(ctx, "initialize", map[string]interface{}{ + "protocolVersion": "2025-03-26", + "capabilities": map[string]interface{}{}, + "clientInfo": map[string]string{"name": "prexus-intelligence", "version": "2.1.0"}, + }) +} + +func (c *Client) Initialized(ctx context.Context) error { + return c.notify(ctx, "notifications/initialized", nil) +} + +func (c *Client) ListTools(ctx context.Context) (json.RawMessage, error) { + return c.call(ctx, "tools/list", map[string]interface{}{}) +} + +func (c *Client) CallTool(ctx context.Context, name string, arguments map[string]interface{}) (json.RawMessage, error) { + return c.call(ctx, "tools/call", map[string]interface{}{"name": name, "arguments": arguments}) +} diff --git a/backend/backend/integrations/conduit/config.go b/backend/backend/integrations/conduit/config.go new file mode 100644 index 0000000..31614aa --- /dev/null +++ b/backend/backend/integrations/conduit/config.go @@ -0,0 +1,48 @@ +package conduit + +import ( + "fmt" + "os" + "strings" +) + +// Config contains only the client-side configuration needed to consume the +// independent Conduit MCP service. Conduit remains an external service; this +// package is the Prexus integration boundary. +type Config struct { + MCPURL string + TokenURL string + Audience string + ClientID string + ClientSecret string + Scope string +} + +func LoadConfig() (Config, error) { + cfg := Config{ + MCPURL: strings.TrimRight(os.Getenv("CONDUIT_MCP_URL"), "/"), + TokenURL: strings.TrimRight(os.Getenv("CONDUIT_TOKEN_URL"), "/"), + Audience: strings.TrimSpace(os.Getenv("CONDUIT_AUDIENCE")), + ClientID: strings.TrimSpace(os.Getenv("CONDUIT_CLIENT_ID")), + ClientSecret: os.Getenv("CONDUIT_CLIENT_SECRET"), + Scope: strings.TrimSpace(os.Getenv("CONDUIT_SCOPE")), + } + + if cfg.MCPURL == "" { + cfg.MCPURL = "https://conduit-mcp-nfmm.onrender.com/mcp" + } + if cfg.TokenURL == "" { + cfg.TokenURL = "https://dev-jf6pbb4exdzatprm.eu.auth0.com/oauth/token" + } + if cfg.Audience == "" { + cfg.Audience = "https://conduit-mcp.onrender.com/mcp" + } + if cfg.Scope == "" { + cfg.Scope = "conduit:read" + } + + if cfg.ClientID == "" || cfg.ClientSecret == "" { + return Config{}, fmt.Errorf("CONDUIT_CLIENT_ID and CONDUIT_CLIENT_SECRET are required") + } + return cfg, nil +}