Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions control-plane/cmd/control-plane/drop_config_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
package main

import "testing"

func TestLoadConfigReadsDropURL(t *testing.T) {
t.Setenv("DROP_URL", "http://drop-external.drop.svc.cluster.local")
cfg, err := loadConfig()
if err != nil {
t.Fatalf("loadConfig: %v", err)
}
if got, want := cfg.dropURL, "http://drop-external.drop.svc.cluster.local"; got != want {
t.Fatalf("dropURL = %q, want %q", got, want)
}
t.Setenv("DROP_URL", "")
if cfg, _ := loadConfig(); cfg.dropURL != "" {
t.Fatalf("dropURL = %q, want off when unset", cfg.dropURL)
}
}
6 changes: 5 additions & 1 deletion control-plane/cmd/control-plane/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@ func main() {
"claude_code_default_model", cfg.claudeCodeDefaultModel,
"claude_code_models", len(cfg.claudeCodeModels),
"repo_access_enabled", cfg.repoAccess.enabled,
"drop_url", cfg.dropURL,
)
if cfg.dataPlaneImage == "" {
// The in-code fallback image has no session agent, so session pods will
Expand All @@ -86,7 +87,8 @@ func main() {
k8s.WithMiniMaxCredentialsSecret(cfg.miniMax.credentialsSecret),
k8s.WithApprovalGatewayURL(cfg.approvalGatewayURL),
k8s.WithSharedVolume(cfg.sharedVolumeStorageClass, cfg.sharedVolumeSize),
k8s.WithRepoAccess(cfg.repoAccess.helperURL(namespace)))
k8s.WithRepoAccess(cfg.repoAccess.helperURL(namespace)),
k8s.WithDropURL(cfg.dropURL))
store := configmap.NewStore(client, namespace)
// Shell I/O (write→stdin, read→scrollback delta) AND checkpoint/restore ride
// the same agent client: it resolves pod name → IP per request and dials the
Expand Down Expand Up @@ -252,6 +254,7 @@ type config struct {
checkpointS3Endpoint string
miniMax miniMaxConfig
repoAccess repoAccessConfig
dropURL string // drop's in-cluster URL for helper pods (docs/prd/drop-file-share.md AC-I2)
}

// GET /config lists only these, so clients never offer a type whose create would fail.
Expand Down Expand Up @@ -361,6 +364,7 @@ func loadConfig() (config, error) {
checkpointS3SessionName: env("CHECKPOINT_S3_SESSION_NAME", ""),
miniMax: loadMiniMaxConfig(),
repoAccess: loadRepoAccessConfig(),
dropURL: env("DROP_URL", ""),
}, nil
}

Expand Down
2 changes: 2 additions & 0 deletions control-plane/internal/adapter/k8s/client_orchestrator.go
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,7 @@ type ClientOrchestrator struct {
claudeCredentialsSecret string // platform Secret referenced by claude-code pods
approvalGatewayURL string // PocketAide base URL handed to approval-gated helper pods
repoAccessURL string // control plane URL for the helper's repo access calls (repo_access.go)
dropURL string // drop's in-cluster URL for the helper's drop_* tools (drop.go)
miniMaxCredentialsSecret string
// AC-F5's per-session shared claim — see shared_volume.go.
sharedVolumeStorageClass string
Expand Down Expand Up @@ -737,6 +738,7 @@ func (o *ClientOrchestrator) helperPodSpec(sessionID, suffix string, workloadTyp
helperVolumes = append(helperVolumes, sharedVolume(sharedClaimName(name)))
}
helperVolumes = o.addRepoAccess(&mcp, helperVolumes)
helperVolumes = o.addDrop(&mcp, helperVolumes)

return &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Expand Down
59 changes: 59 additions & 0 deletions control-plane/internal/adapter/k8s/drop.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
package k8s

import (
"strings"

corev1 "k8s.io/api/core/v1"
)

// drop 파일 전달(docs/prd/drop-file-share.md AC-I2): 헬퍼 파드의 세션 MCP 가 drop 에 자기
// 서비스 어카운트로 닿는 자리. 워크로드 파드는 drop 에 닿지 않고(AC-F2 그대로) 토큰도 받지 않는다.
const (
// data-plane/cmd/agent 의 dropURLEnv·dropTokenFileEnv 와 맞춘다.
DropURLEnvVar = "DROP_URL"
DropTokenFileEnvVar = "DROP_TOKEN_FILE"
// DropTokenAudience 는 drop 의 SA_TOKEN_AUDIENCE 와 같아야 한다. 다른 audience 토큰처럼
// apiserver 의 기본 audience 가 아니므로 drop 이 이 토큰으로 apiserver 를 부를 수 없다.
DropTokenAudience = "drop"
dropTokenVolume = "drop-token"
dropTokenDir = "/var/run/secrets/drop"
dropTokenPath = "token"
dropTokenExpirySec = int64(3600)
)

// WithDropURL 은 헬퍼 파드에 drop 의 클러스터 안 주소를 준다. 비우면 기능이 꺼진다 — 토큰을
// 투영하지 않고 세션 MCP 의 도구 목록에 drop 도구가 오르지 않는다.
func WithDropURL(url string) Option {
return func(o *ClientOrchestrator) {
o.dropURL = strings.TrimRight(strings.TrimSpace(url), "/")
}
}

func (o *ClientOrchestrator) dropEnabled() bool { return o.dropURL != "" }

// addDrop 은 세션 MCP 컨테이너에만 drop 주소와 audience drop 의 projected 토큰을 준다.
func (o *ClientOrchestrator) addDrop(mcp *corev1.Container, volumes []corev1.Volume) []corev1.Volume {
if !o.dropEnabled() {
return volumes
}
mcp.Env = append(mcp.Env,
corev1.EnvVar{Name: DropURLEnvVar, Value: o.dropURL},
corev1.EnvVar{Name: DropTokenFileEnvVar, Value: dropTokenDir + "/" + dropTokenPath},
)
mcp.VolumeMounts = append(mcp.VolumeMounts, corev1.VolumeMount{
Name: dropTokenVolume, MountPath: dropTokenDir, ReadOnly: true,
})
expiry := dropTokenExpirySec
return append(volumes, corev1.Volume{
Name: dropTokenVolume,
VolumeSource: corev1.VolumeSource{Projected: &corev1.ProjectedVolumeSource{
Sources: []corev1.VolumeProjection{{
ServiceAccountToken: &corev1.ServiceAccountTokenProjection{
Audience: DropTokenAudience,
ExpirationSeconds: &expiry,
Path: dropTokenPath,
},
}},
}},
})
}
127 changes: 127 additions & 0 deletions control-plane/test/drop_orchestrator_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
//go:build integration

// drop 파일 전달이 켜졌을 때의 헬퍼 파드 모양(docs/prd/drop-file-share.md AC-I2).
package integration_test

import (
"context"
"testing"

corev1 "k8s.io/api/core/v1"

"github.com/dlddu/session-platform/control-plane/internal/adapter/k8s"
"github.com/dlddu/session-platform/control-plane/internal/session"
)

const dropURL = "http://drop-external.drop.svc.cluster.local"

func startWithDrop(t *testing.T, sessionID string, workloadType session.WorkloadType, on bool) podSet {
t.Helper()
var opts []k8s.Option
if on {
opts = append(opts, k8s.WithDropURL(" "+dropURL+"/ "))
}
orch, cs := newMiniMaxOrchestrator(t, append(opts,
k8s.WithWorkloadImage(session.WorkloadTypeApprovalGated, approvalGatedImage))...)
started, err := orch.Start(context.Background(), sessionID, k8s.WorkloadSpec{Type: workloadType})
if err != nil {
t.Fatalf("start %s session: %v", workloadType, err)
}
return newPodSet(t, listPods(t, cs), started)
}

func dropTokenVolume(pod corev1.Pod) (string, *corev1.ServiceAccountTokenProjection) {
for _, v := range pod.Spec.Volumes {
if v.Projected == nil {
continue
}
for _, src := range v.Projected.Sources {
if sa := src.ServiceAccountToken; sa != nil && sa.Audience == k8s.DropTokenAudience {
return v.Name, sa
}
}
}
return "", nil
}

func TestDrop_OffLeavesHelperPodsAsTheyWere(t *testing.T) {
for _, wt := range []session.WorkloadType{session.WorkloadTypeApprovalGated, session.WorkloadTypeMiniMaxCode} {
set := startWithDrop(t, "dr0"+string(wt[0]), wt, false)
if name, _ := dropTokenVolume(set.helper); name != "" {
t.Errorf("%s: drop token projected while DROP_URL is unset", wt)
}
mcp := container(t, set.helper, k8s.SessionMCPContainerName)
for _, env := range []string{k8s.DropURLEnvVar, k8s.DropTokenFileEnvVar} {
if _, ok := envValue(mcp, env); ok {
t.Errorf("%s: session MCP has %s while DROP_URL is unset", wt, env)
}
}
}
}

func TestDrop_SessionMCPAloneHoldsTheDropToken(t *testing.T) {
for _, wt := range []session.WorkloadType{session.WorkloadTypeApprovalGated, session.WorkloadTypeMiniMaxCode} {
set := startWithDrop(t, "dr1"+string(wt[0]), wt, true)
mcp := container(t, set.helper, k8s.SessionMCPContainerName)
if got, _ := envValue(mcp, k8s.DropURLEnvVar); got != dropURL {
t.Errorf("%s: %s = %q, want %q", wt, k8s.DropURLEnvVar, got, dropURL)
}
volume, sa := dropTokenVolume(set.helper)
if sa == nil {
t.Fatalf("%s: no projected token for audience %q", wt, k8s.DropTokenAudience)
}
if sa.ExpirationSeconds == nil || *sa.ExpirationSeconds > 3600 {
t.Errorf("%s: token expiry = %v, want at most an hour", wt, sa.ExpirationSeconds)
}
tokenFile, _ := envValue(mcp, k8s.DropTokenFileEnvVar)
if tokenFile != "/var/run/secrets/drop/token" {
t.Errorf("%s: %s = %q", wt, k8s.DropTokenFileEnvVar, tokenFile)
}
if set.helper.Spec.AutomountServiceAccountToken == nil || *set.helper.Spec.AutomountServiceAccountToken {
t.Errorf("%s: helper pod automounts an apiserver token", wt)
}
if set.helper.Spec.ServiceAccountName != k8s.SessionMCPServiceAccountName {
t.Errorf("%s: helper runs as %q — drop's allow list names session-mcp", wt, set.helper.Spec.ServiceAccountName)
}
for _, pod := range []corev1.Pod{set.workload, set.helper} {
for _, c := range pod.Spec.Containers {
isMCP := pod.Name == set.helper.Name && c.Name == k8s.SessionMCPContainerName
mounted := false
for _, m := range c.VolumeMounts {
if m.Name == volume && pod.Name == set.helper.Name {
mounted = true
if !m.ReadOnly || tokenFile != m.MountPath+"/"+sa.Path {
t.Errorf("%s: mount %+v does not serve %s read-only", wt, m, tokenFile)
}
}
}
if mounted != isMCP {
t.Errorf("%s: container %s/%s mounts the drop token = %v", wt, pod.Name, c.Name, mounted)
}
for _, env := range []string{k8s.DropURLEnvVar, k8s.DropTokenFileEnvVar} {
if _, ok := envValue(c, env); ok && !isMCP {
t.Errorf("%s: container %s/%s receives %s", wt, pod.Name, c.Name, env)
}
}
}
}
}
}

// drop 은 헬퍼 파드가 나가는 곳이라 워크로드 파드의 경계는 움직이지 않는다(AC-F2 그대로).
func TestDrop_WorkloadBoundaryIsUnchanged(t *testing.T) {
orch, cs := newMiniMaxOrchestrator(t,
k8s.WithDropURL(dropURL),
k8s.WithWorkloadImage(session.WorkloadTypeApprovalGated, approvalGatedImage))
started, err := orch.Start(context.Background(), "dr2", k8s.WorkloadSpec{Type: session.WorkloadTypeApprovalGated})
if err != nil {
t.Fatal(err)
}
set := newPodSet(t, listPods(t, cs), started)
policies := listNetworkPolicies(t, cs)
egress := policies[set.helper.Name+"-workload-egress"]
if len(egress.Spec.Egress) != 2 {
t.Fatalf("egress rules = %d, want 2", len(egress.Spec.Egress))
}
assertPorts(t, "workload egress to helper", egress.Spec.Egress[1].Ports, 8091, 8092)
}
116 changes: 116 additions & 0 deletions control-plane/test/e2e_drop_1_round_trip_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
//go:build e2e

// 검증 시나리오: drop-file-share.md#시나리오 1
//
// docs/prd/drop-file-share.md AC-I1~I3 를 배포된 SUT 에서: 에이전트가 공유 볼륨에 둔 파일이
// 실물 drop(deploy/e2e-drop.yaml)으로 올라가고, 같은 보관함이 다시 공유 볼륨으로 내려온다.
// 도구 호출은 공급자 대역의 e2e-tool 지시자로 실 CLI 가 세션 MCP 에 낸다(f3 과 같은 진입).
package e2e_test

import (
"encoding/json"
"net/http"
"strings"
"testing"
"time"
)

const (
dropInClusterURL = "http://drop.drop.svc.cluster.local"
dropUploadName = "mcp__session-platform-session-mcp__drop_upload"
dropDownloadName = "mcp__session-platform-session-mcp__drop_download"
// CLI 기동 + MCP 왕복 + drop 의 TokenReview. 승인을 기다리지 않으므로 f3 보다 짧아도 되지만
// 첫 invocation 의 CLI 기동이 러너에 따라 길다.
dropToolBudget = 150 * time.Second
)

func TestDropRoundTrip_FilesLeaveAndReturnThroughTheSharedVolume(t *testing.T) {
cs, cfg, ok := kubeClient(t)
if !ok {
t.Skip("no cluster access; the file is placed and read back through exec")
}
ns := sessionNamespace()
status, s := createTyped(t, map[string]any{"name": uniqueName(t), "workloadType": "approval-gated"})
if status != http.StatusCreated {
t.Fatalf("create approval-gated session: status=%d", status)
}
t.Cleanup(func() {
if resp, raw := do(t, http.MethodDelete, "/api/v1/sessions/"+s.ID, nil); resp.StatusCode >= 400 {
t.Logf("cleanup delete %s: status=%d body=%s", s.ID, resp.StatusCode, raw)
}
})
helpers := helperPodsFor(t, cs, ns, s.ID)
if len(helpers) != 1 {
t.Fatalf("session %s has %d helper pods, want 1", s.ID, len(helpers))
}
helper := helpers[0].Name
shared := strings.TrimSpace(f4Sh(t, cs, cfg, ns, s.Pod, workloadContainer, `printf %s "${SESSION_SHARED_DIR-}"`))
if shared == "" {
t.Fatal("the workload container has no SESSION_SHARED_DIR; the kind overlay names a ReadWriteMany class, so this is a wiring fault")
}

// 에이전트 쪽에서 파일을 둔다 — 도구 인자에는 이 경로만 실린다(AC-I1).
marker := "drop-e2e-contents-" + s.ID
f4Sh(t, cs, cfg, ns, s.Pod, workloadContainer,
`set -eu; mkdir -p "$1/out"; printf '%s' "$2" > "$1/out/hello.txt"`, shared, marker)

id := "sp-" + s.ID[:16]
password := "pw-" + s.ID[16:28]
upload, _ := json.Marshal(map[string]any{"paths": []string{"out/hello.txt"}, "id": id, "password": password, "days": 1})
writeShell(t, s.ID, "e2e-tool:"+dropUploadName+":"+string(upload))

// 보관함은 drop 자신에게 묻는다 — 받는 사람이 쓰는 것과 같은 받기 API 로, 세션 MCP 를 거치지 않고.
listing := dropEventually(t, func() (string, bool) {
out := f4Sh(t, cs, cfg, ns, helper, sessionMCPContainer,
`curl -sS --max-time 20 -w '\n%{http_code}' -u "$2:$3" -H 'Accept: application/json' "$1/get/$2"`,
dropInClusterURL, id, password)
i := strings.LastIndex(out, "\n")
if i < 0 {
return out, false
}
return out[:i], strings.TrimSpace(out[i+1:]) == "200"
}, "drop never held the uploaded drop "+id)
var got struct {
Files []struct {
Name string `json:"name"`
Size int64 `json:"size"`
} `json:"files"`
}
if err := json.Unmarshal([]byte(listing), &got); err != nil {
t.Fatalf("decode drop's listing: %v raw=%s", err, listing)
}
if len(got.Files) != 1 || got.Files[0].Name != "hello.txt" || got.Files[0].Size != int64(len(marker)) {
t.Fatalf("drop lists %+v, want hello.txt of %d bytes", got.Files, len(marker))
}
if content := f4Sh(t, cs, cfg, ns, helper, sessionMCPContainer,
`curl -sS --fail --max-time 20 -u "$2:$3" "$1/get/$2/f/0"`, dropInClusterURL, id, password); content != marker {
t.Fatalf("drop serves %q, want the bytes the agent wrote", content)
}

download, _ := json.Marshal(map[string]any{"id": id, "password": password, "dest": "in"})
writeShell(t, s.ID, "e2e-tool:"+dropDownloadName+":"+string(download))
dropEventually(t, func() (string, bool) {
out := f4Sh(t, cs, cfg, ns, s.Pod, workloadContainer, `cat "$1/in/hello.txt" 2>/dev/null || true`, shared)
return out, out == marker
}, "the downloaded file never reached the workload's shared volume")

// AC-I3: 승인을 기다린 적이 없다. 두 호출 모두 대기 마커 없이 끝났다.
if buf := readShellAt(t, s.ID, 0).Payload; strings.Contains(buf, "[session-platform: awaiting approval") {
t.Fatalf("a drop tool waited on the approval gate: %q", buf)
}
}

func dropEventually(t *testing.T, probe func() (string, bool), why string) string {
t.Helper()
deadline := time.Now().Add(dropToolBudget)
for {
out, ok := probe()
if ok {
return out
}
if time.Now().After(deadline) {
t.Fatalf("%s within %s (last probe: %q)", why, dropToolBudget, out)
}
time.Sleep(2 * time.Second)
}
}
5 changes: 3 additions & 2 deletions data-plane/cmd/agent/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -174,9 +174,10 @@ func main() {
k3sProxyMCP = mcpURL
}
}
handler = sessionMCPRoutes(logger, newSessionMCPConfig(gateway, shared, repoAccess, k3sProxyMCP))
drop := newDropClient(os.Getenv(dropURLEnv), os.Getenv(dropTokenFileEnv))
handler = sessionMCPRoutes(logger, newSessionMCPConfig(gateway, shared, repoAccess, k3sProxyMCP, drop))
logger.Info("session MCP started", "addr", addr, "gated", gateway != nil, "shared_dir", shared,
"repo_access", repoAccess != nil, "k3s_mcp_proxy", k3sProxyMCP != "")
"repo_access", repoAccess != nil, "k3s_mcp_proxy", k3sProxyMCP != "", "drop", drop != nil)
case workloadCredentialProxy:
placement, err := credentialProxyPlacementFromEnv()
if err != nil {
Expand Down
Loading
Loading