Add project-scoped Codex execution over a private socket

2026-11-05 02:29:49 +08:00
parent 4d5e6e2512
commit 9e5d805f5a
7 changed files with 426 additions and 3 deletions
@@ -0,0 +1,143 @@
+package codex
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"crypto/subtle"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+)
+
+const (
+	maxRequest = 32 << 10
+	maxOutput  = 512 << 10
+	timeout    = 90 * time.Second
+)
+
+type task struct {
+	Prompt string `json:"prompt"`
+}
+
+func writeJSON(w http.ResponseWriter, status int, value any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("Cache-Control", "no-store")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(value)
+}
+
+func fail(w http.ResponseWriter, status int, code string) {
+	writeJSON(w, status, map[string]string{"error": code})
+}
+
+func readTask(w http.ResponseWriter, r *http.Request, limit int64) (task, bool) {
+	var input task
+	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, limit))
+	decoder.DisallowUnknownFields()
+	err := decoder.Decode(&input)
+	if err == nil {
+		err = decoder.Decode(new(any))
+		if err == io.EOF && strings.TrimSpace(input.Prompt) != "" {
+			return input, true
+		}
+	}
+	var tooLarge *http.MaxBytesError
+	if errors.As(err, &tooLarge) {
+		fail(w, 413, "request_too_large")
+	} else {
+		fail(w, 400, "invalid_task")
+	}
+	return task{}, false
+}
+
+// Register enables the public endpoint only when both configuration files are supplied.
+// Each credential digest belongs to exactly one project.
+func Register(mux *http.ServeMux, projects map[string]string, socket, grantsFile string) error {
+	if socket == "" && grantsFile == "" {
+		return nil
+	}
+	if socket == "" || grantsFile == "" {
+		return errors.New("Codex requires both a socket and a grants file")
+	}
+	transport := &http.Transport{
+		DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+			return (&net.Dialer{}).DialContext(ctx, "unix", socket)
+		},
+		DisableKeepAlives: true,
+	}
+	client := &http.Client{
+		Transport:     transport,
+		Timeout:       timeout + 2*time.Second,
+		CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
+	}
+	mux.HandleFunc("POST /projects/{project}/codex/exec", func(w http.ResponseWriter, r *http.Request) {
+		scheme, token, _ := strings.Cut(r.Header.Get("Authorization"), " ")
+		key, err := hex.DecodeString(token)
+		if !strings.EqualFold(scheme, "Bearer") || err != nil || len(key) != 32 {
+			w.Header().Set("WWW-Authenticate", "Bearer")
+			fail(w, 401, "unauthorized")
+			return
+		}
+		data, err := os.ReadFile(grantsFile)
+		var grants map[string]string
+		if err != nil || json.Unmarshal(data, &grants) != nil {
+			fail(w, 503, "codex_unavailable")
+			return
+		}
+		hash := sha256.Sum256(key)
+		var scope string
+		for digest, project := range grants {
+			value, err := hex.DecodeString(digest)
+			if err == nil && subtle.ConstantTimeCompare(hash[:], value) == 1 {
+				scope = project
+			}
+		}
+		if scope == "" {
+			w.Header().Set("WWW-Authenticate", "Bearer")
+			fail(w, 401, "unauthorized")
+			return
+		}
+		project := r.PathValue("project")
+		if _, exists := projects[project]; !exists || scope != project {
+			fail(w, 403, "forbidden")
+			return
+		}
+		input, ok := readTask(w, r, maxRequest)
+		if !ok {
+			return
+		}
+		body, _ := json.Marshal(input)
+		req, err := http.NewRequestWithContext(r.Context(), http.MethodPost, "http://codex/exec", bytes.NewReader(body))
+		if err != nil {
+			fail(w, 502, "codex_unavailable")
+			return
+		}
+		resp, err := client.Do(req)
+		if err != nil {
+			if errors.Is(err, context.DeadlineExceeded) {
+				fail(w, 504, "codex_timeout")
+			} else {
+				fail(w, 502, "codex_unavailable")
+			}
+			return
+		}
+		defer resp.Body.Close()
+		data, err = io.ReadAll(io.LimitReader(resp.Body, maxOutput+1))
+		if err != nil || len(data) > maxOutput || !json.Valid(data) {
+			fail(w, 502, "invalid_codex_response")
+			return
+		}
+		if retry := resp.Header.Get("Retry-After"); retry != "" {
+			w.Header().Set("Retry-After", retry)
+		}
+		writeJSON(w, resp.StatusCode, json.RawMessage(data))
+	})
+	return nil
+}
@@ -0,0 +1,69 @@
+package codex
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestProjectAuthorizationAndForwarding(t *testing.T) {
+	dir := t.TempDir()
+	socket := filepath.Join(dir, "worker.sock")
+	listener, err := net.Listen("unix", socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	worker := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/exec" || r.Header.Get("Authorization") != "" {
+			fail(w, 500, "unexpected_worker_request")
+			return
+		}
+		input, ok := readTask(w, r, 6*maxRequest)
+		if ok {
+			writeJSON(w, 200, map[string]string{"result": input.Prompt})
+		}
+	})}
+	go worker.Serve(listener)
+	t.Cleanup(func() { _ = worker.Close() })
+
+	key := make([]byte, 32)
+	token := hex.EncodeToString(key)
+	grants := filepath.Join(dir, "grants.json")
+	writeGrants := func(data string) {
+		t.Helper()
+		if err := os.WriteFile(grants, []byte(data), 0600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	writeGrants(fmt.Sprintf(`{"%x":"notes"}`, sha256.Sum256(key)))
+	mux := http.NewServeMux()
+	if err := Register(mux, map[string]string{"notes": "", "board": ""}, socket, grants); err != nil {
+		t.Fatal(err)
+	}
+	check := func(project, credential, body string, status int, response string) {
+		t.Helper()
+		r := httptest.NewRequest("POST", "/projects/"+project+"/codex/exec", strings.NewReader(body))
+		if credential != "" {
+			r.Header.Set("Authorization", "Bearer "+credential)
+		}
+		w := httptest.NewRecorder()
+		mux.ServeHTTP(w, r)
+		if w.Code != status || !strings.Contains(w.Body.String(), response) {
+			t.Fatalf("got %d %s, want %d containing %q", w.Code, w.Body.String(), status, response)
+		}
+	}
+	check("notes", "", `{"prompt":"hello"}`, 401, "unauthorized")
+	check("board", token, `{"prompt":"hello"}`, 403, "forbidden")
+	check("notes", token, `{"prompt":"hello","sandbox":"danger"}`, 400, "invalid_task")
+	check("notes", token, `{"prompt":"`+strings.Repeat("x", maxRequest)+`"}`, 413, "request_too_large")
+	check("notes", token, `{"prompt":"hello"}`, 200, `"result":"hello"`)
+	writeGrants("{}")
+	check("notes", token, `{"prompt":"hello"}`, 401, "unauthorized")
+}
@@ -0,0 +1,134 @@
+package codex
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"strings"
+	"syscall"
+	"time"
+)
+
+const executable = "/opt/codex/node_modules/.bin/codex"
+
+var errOutputLimit = errors.New("Codex output limit exceeded")
+
+type outputBuffer struct {
+	bytes.Buffer
+	exceeded bool
+}
+
+func (b *outputBuffer) Write(p []byte) (int, error) {
+	if len(p) > maxOutput-b.Len() {
+		b.exceeded = true
+		return 0, errOutputLimit
+	}
+	return b.Buffer.Write(p)
+}
+
+func ServeWorker(socket string) error {
+	syscall.Umask(0077)
+	if err := os.Remove(socket); err != nil && !errors.Is(err, os.ErrNotExist) {
+		return err
+	}
+	listener, err := net.Listen("unix", socket)
+	if err != nil {
+		return err
+	}
+	defer listener.Close()
+	if err := os.Chmod(socket, 0600); err != nil {
+		return err
+	}
+	slot := make(chan struct{}, 1)
+	mux := http.NewServeMux()
+	mux.HandleFunc("POST /exec", func(w http.ResponseWriter, r *http.Request) {
+		select {
+		case slot <- struct{}{}:
+			defer func() { <-slot }()
+		default:
+			w.Header().Set("Retry-After", "5")
+			fail(w, 429, "codex_busy")
+			return
+		}
+		input, ok := readTask(w, r, 6*maxRequest)
+		if ok {
+			execute(w, r, input.Prompt)
+		}
+	})
+	server := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: timeout + 15*time.Second}
+	return server.Serve(listener)
+}
+
+func execute(w http.ResponseWriter, r *http.Request, prompt string) {
+	ctx, cancel := context.WithTimeout(r.Context(), timeout)
+	defer cancel()
+	cmd := exec.CommandContext(ctx, executable, "--ask-for-approval", "never",
+		"--disable", "shell_tool", "--disable", "apps", "--disable", "plugins",
+		"--disable", "multi_agent", "--disable", "hooks", "--disable", "memories",
+		"-c", `web_search="disabled"`, "exec", "--json", "--sandbox", "read-only", "--skip-git-repo-check", "--ephemeral", "-")
+	cmd.Dir = "/work"
+	cmd.Env = []string{"PATH=/usr/local/bin:/usr/bin:/bin", "HOME=/home/node", "CODEX_HOME=/home/node/.codex", "SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt"}
+	cmd.Stdin = strings.NewReader(prompt)
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	cmd.Cancel = func() error {
+		err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
+		if errors.Is(err, syscall.ESRCH) {
+			return os.ErrProcessDone
+		}
+		return err
+	}
+	cmd.WaitDelay = 2 * time.Second
+	var output outputBuffer
+	cmd.Stdout, cmd.Stderr = &output, io.Discard
+	err := cmd.Run()
+	switch {
+	case output.exceeded:
+		fail(w, 502, "codex_output_too_large")
+	case errors.Is(ctx.Err(), context.DeadlineExceeded):
+		fail(w, 504, "codex_timeout")
+	case err != nil:
+		fail(w, 502, "codex_failed")
+	default:
+		var result string
+		var completed bool
+		for _, line := range bytes.Split(bytes.TrimSpace(output.Bytes()), []byte("\n")) {
+			var event struct {
+				Type string `json:"type"`
+				Item struct {
+					Type string `json:"type"`
+					Text string `json:"text"`
+				} `json:"item"`
+			}
+			if json.Unmarshal(line, &event) != nil {
+				fail(w, 502, "invalid_codex_response")
+				return
+			}
+			if event.Type == "item.completed" && event.Item.Type == "agent_message" {
+				result = event.Item.Text
+			}
+			if event.Type == "turn.failed" || event.Type == "error" {
+				fail(w, 502, "codex_failed")
+				return
+			}
+			if event.Type == "turn.completed" {
+				completed = true
+			}
+		}
+		if !completed || result == "" {
+			fail(w, 502, "codex_failed")
+			return
+		}
+		response, _ := json.Marshal(map[string]string{"result": result})
+		if len(response) >= maxOutput {
+			fail(w, 502, "codex_output_too_large")
+			return
+		}
+		writeJSON(w, 200, json.RawMessage(response))
+	}
+}