diff --git a/src/frontend/apiserver/backend/convert_pure.go b/src/frontend/apiserver/backend/convert_pure.go index 8eb9469f..feb0cf8d 100644 --- a/src/frontend/apiserver/backend/convert_pure.go +++ b/src/frontend/apiserver/backend/convert_pure.go @@ -549,8 +549,8 @@ type rawJobData struct { submit *rawSubmitData execHosts []rawHostSlot pendReason *rawPendReason - runUsage *rawRunUsage // nil until the job's first usage report - pidInfo *rawPidInfo // nil until the worker's start report + runUsage *rawRunUsage // nil until the job's first usage report + pidInfo *rawPidInfo // nil until the worker's start report } // rawWorkerData mirrors the C xce_worker_t fields used by the Go model. @@ -606,6 +606,24 @@ func interactiveURLFromExdata(exdata string) string { return obj.URL } +// x11CookieFromExdata extracts the per-job X11 auth cookie of an -X job +// from the same start-report exdata (the worker packs it as +// {"x11cookie": ...} next to the x11:// URL). The xcecli -X client needs it +// to register the session's MIT-MAGIC-COOKIE-1 with the user's Xauthority +// before the app's first X connection is relayed through. +func x11CookieFromExdata(exdata string) string { + if exdata == "" { + return "" + } + var obj struct { + Cookie string `json:"x11cookie"` + } + if err := json.Unmarshal([]byte(exdata), &obj); err != nil { + return "" + } + return obj.Cookie +} + // pidInfoJSONB marshals a start report's pids/pgids into the PIDInfo Jsonb // the API exposes (processJobData parses the JSONB string before the row // leaves the API, so clients see an object with pids/pgids). Shared by @@ -730,11 +748,13 @@ func rawJobToModel(r rawJobData) *models.JobInfo { // string before the row leaves the API, so clients see an object // with pids/pgids. The interactive URL is promoted to its own // top-level field — it is what xcecli's waitForInteractiveURL and - // the --browse flow consume. + // the --browse flow consume. An -X job's auth cookie rides the same + // report and gets the same treatment. if jb, ok := pidInfoJSONB(p.pids, p.pgids); ok { info.PIDInfo = jb } info.InteractiveJobURL = interactiveURLFromExdata(p.exdata) + info.X11Cookie = x11CookieFromExdata(p.exdata) } return info } diff --git a/src/frontend/apiserver/backend/convert_test.go b/src/frontend/apiserver/backend/convert_test.go index b10ccc05..11ea252f 100644 --- a/src/frontend/apiserver/backend/convert_test.go +++ b/src/frontend/apiserver/backend/convert_test.go @@ -455,7 +455,7 @@ func TestRawJobToModel_PendReasonDetailOnlyPromotesName(t *testing.T) { jobID: 9, state: 1, // JOB_STAT_PEND pendReason: &rawPendReason{ - MainReason: 0, // detail-only: no explicit main reason was set + MainReason: 0, // detail-only: no explicit main reason was set HostReasons: []rawHostReason{{Host: "t2", Reason: 2}, {Host: "t1", Reason: 3}}, ResReasons: []rawResReason{{Resource: "slots", Host: "", Reason: 4}}, }, @@ -608,6 +608,57 @@ func TestRawJobToModel_NilPidInfo(t *testing.T) { } } +// An -X job's start report carries the per-job X11 auth cookie next to the +// x11:// URL; both must surface on the job row (the cookie is what the +// xcecli -X client registers with the user's Xauthority). +func TestRawJobToModel_X11CookieFromPidInfoExdata(t *testing.T) { + r := rawJobData{ + jobID: 14, + state: 2, + pidInfo: &rawPidInfo{ + pids: []int{4321}, + exdata: `{"url": "x11://10.0.1.5:6100", "x11": 1, "x11cookie": "aabbccdd00112233445566778899eeff"}`, + }, + } + + info := rawJobToModel(r) + + if want := "x11://10.0.1.5:6100"; info.InteractiveJobURL != want { + t.Errorf("InteractiveJobURL: got %q, want %q", info.InteractiveJobURL, want) + } + if want := "aabbccdd00112233445566778899eeff"; info.X11Cookie != want { + t.Errorf("X11Cookie: got %q, want %q", info.X11Cookie, want) + } +} + +// Non-X jobs never carry a cookie: ttyd/VNC exdata must not leak into +// X11Cookie (xcecli refuses to attach -X without one, so a wrong cookie +// would break the auth model silently). +func TestRawJobToModel_X11CookieAbsentForOtherJobs(t *testing.T) { + cases := []struct { + name string + exdata string + }{ + {"ttyd attach", `{"url": "http://10.0.1.5:9000"}`}, + {"vnc xjob", `{"url": "10.0.1.5::5901"}`}, + {"x11 without cookie key", `{"url": "x11://10.0.1.5:6100", "x11": 1}`}, + {"malformed", "not json"}, + {"empty", ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + info := rawJobToModel(rawJobData{ + jobID: 15, + state: 2, + pidInfo: &rawPidInfo{pids: []int{1}, exdata: tc.exdata}, + }) + if info.X11Cookie != "" { + t.Errorf("X11Cookie = %q, want empty for exdata %q", info.X11Cookie, tc.exdata) + } + }) + } +} + func TestRawJobToModel_Timestamps(t *testing.T) { r := rawJobData{ jobID: 1, diff --git a/src/frontend/apiserver/backend/handler.go b/src/frontend/apiserver/backend/handler.go index b72073b2..d484f839 100644 --- a/src/frontend/apiserver/backend/handler.go +++ b/src/frontend/apiserver/backend/handler.go @@ -142,6 +142,7 @@ func (h *DefaultHandler) OnJobPidInfo(jobid uint64, pidInfo *rawPidInfo) { existing.PIDInfo = jb } existing.InteractiveJobURL = interactiveURLFromExdata(pidInfo.exdata) + existing.X11Cookie = x11CookieFromExdata(pidInfo.exdata) } store.Jobs.Put(jobid, existing) DefaultBroker.Publish(Event{Type: EventJobUpdated, JobID: jobid, diff --git a/src/frontend/apiserver/controllers/job_xattach.go b/src/frontend/apiserver/controllers/job_xattach.go new file mode 100644 index 00000000..1d56767d --- /dev/null +++ b/src/frontend/apiserver/controllers/job_xattach.go @@ -0,0 +1,261 @@ +package controllers + +import ( + "fmt" + "net" + "net/http" + "strings" + "sync" + "time" + + "apiserver/backend" + "apiserver/common" + "apiserver/store" + + logger "apiserver/utils" + + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" + "go.uber.org/zap" +) + +// Interactive XJOB X11-forwarding attach (websocket relay). +// +// An -X job (interactive=2 + exdata x11=1) runs its command with DISPLAY +// pointed at a forwarding port on the worker (inter_job_utils.c, 6100-6199): +// the app's X connections arrive there, and this relay bridges each of them +// to one websocket so the app's windows render on the user's own X server, +// ssh -X style. Worker ports are typically unreachable from user clients in +// production, so the apiserver sits in between exactly as for ttyd attach. +// +// Unlike the ttyd attach, the worker side is a RAW TCP listener -- no +// websocket upgrade, no subprotocol, no ttyd framing. One websocket +// connection carries one X connection byte-for-byte in binary frames +// (X11 needs no framing of its own; text frames are ignored). Close codes +// match the ttyd attach: 4000 = cannot reach the forwarding port, 4001 = +// the session (or its job) ended. Auth and job ownership are checked once +// at relay setup; the worker port is never exposed to the user. + +const ( + // xattachDialTimeout bounds the apiserver→worker TCP dial. + xattachDialTimeout = 10 * time.Second + + // xattachWriteTimeout bounds any single write on either hop. + xattachWriteTimeout = 10 * time.Second +) + +// xattachUpgrader upgrades the client-facing hop (same shape as the ttyd +// attach: the route sits behind JWT auth, so CheckOrigin can stay open). +var xattachUpgrader = websocket.Upgrader{ + ReadBufferSize: 32 * 1024, + WriteBufferSize: 32 * 1024, + CheckOrigin: func(r *http.Request) bool { return true }, +} + +// XAttachJob relays the caller's websocket to one X connection of an -X +// job's forwarding port. GET /xce/v1/jobs/:jobids/xattach (websocket +// upgrade). Auth mirrors AttachJob: Authorization header or ?token= query. +func (j JobController) XAttachJob(ctx *gin.Context) { + logger.Debug("Enter XAttachJob() function from ", zap.String("client_ip", ctx.ClientIP())) + + token := ctx.GetHeader("Authorization") + if token == "" { + if qt := ctx.Query("token"); qt != "" { + token = "Bearer " + qt + } + } + logonUser, err := GetUsernameFromToken(token) + if err != nil { + ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) + return + } + + jobInfo, err := parseJobIDParam(ctx.Param("jobids")) + if err != nil { + ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) + return + } + job, ok := store.Jobs.Get(uint64(jobInfo.Unique_JobID)) + if !ok { + ReturnError(ctx, common.ERROR_NOT_FOUND_CODE, common.ERROR_JOB_NOT_FOUND_MSG, nil) + return + } + + // Same ownership rule as the ttyd attach: the session is job control. + if job.SubmitUser != logonUser { + scope, perr := userJobScope(logonUser, "job:control") + if perr != nil || !scope.covers(job.SubmitUser) { + ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) + return + } + } + + if job.InteractiveJobURL == "" { + ReturnError(ctx, common.INVALID_PARAMS_CODE, + fmt.Sprintf("job <%s> has no interactive session URL (not an interactive job, not running, or started before URL reporting)", + backend.FormatJobID(job.Unique_JobID)), nil) + return + } + workerAddr, err := x11ForwardAddr(job.InteractiveJobURL) + if err != nil { + ReturnError(ctx, common.INVALID_PARAMS_CODE, "invalid X11 session URL: "+err.Error(), nil) + return + } + + // Upgrade the client hop. After this point errors go out as websocket + // close frames, not HTTP responses. + clientConn, err := xattachUpgrader.Upgrade(ctx.Writer, ctx.Request, nil) + if err != nil { + logger.Error("xattach: client upgrade failed", zap.Error(err)) + return + } + defer clientConn.Close() + + workerConn, err := net.DialTimeout("tcp", workerAddr, xattachDialTimeout) + if err != nil { + logger.Warn("xattach: worker dial failed", zap.String("addr", workerAddr), zap.Error(err)) + attachClose(clientConn, attachCloseJobGone, "cannot reach the X11 forwarding port on the worker: "+err.Error()) + return + } + defer workerConn.Close() + + logger.Info("xattach: relaying X11 connection", + zap.Int64("jobid", job.Unique_JobID), zap.String("user", logonUser), + zap.String("worker_addr", workerAddr)) + + xattachRelay(uint64(job.Unique_JobID), clientConn, workerConn) + logger.Info("xattach: relay finished", zap.Int64("jobid", job.Unique_JobID)) +} + +// x11ForwardAddr turns the reported session URL (x11://host:port) into the +// worker TCP address to relay to. +func x11ForwardAddr(raw string) (string, error) { + raw = strings.TrimSpace(raw) + if !strings.HasPrefix(raw, "x11://") { + return "", fmt.Errorf("not an X11 forwarding URL: %q", raw) + } + addr := strings.TrimPrefix(raw, "x11://") + if addr == "" || strings.Contains(addr, "/") { + return "", fmt.Errorf("malformed X11 forwarding URL %q", raw) + } + return addr, nil +} + +// xattachRelay copies bytes between the websocket and the worker TCP +// connection until either side goes away. The job's terminal state ends the +// relay regardless (EventJobFinished watcher, same as the ttyd attach). +func xattachRelay(jobID uint64, client *websocket.Conn, worker net.Conn) { + var wg sync.WaitGroup + wg.Add(2) + done := make(chan struct{}) + // teardown closes both hops once; each pump's blocking read unblocks on + // the close of the socket it is reading, so one close ends both pumps. + // Closing must happen the moment either pump exits -- waiting for + // wg.Wait() here would deadlock (the remaining pump blocks in a read + // nothing else would ever interrupt). The eventJobFinished watcher and + // the late teardown below also call Close; net.Conn and gorilla + // websocket Close are both idempotent. + var closeOnce sync.Once + teardown := func() { + closeOnce.Do(func() { + close(done) + client.Close() + worker.Close() + }) + } + + go func() { + defer wg.Done() + defer teardown() + if !xattachPumpTCP(worker, client, done) { + // worker side ended on its own: the X connection (or its job) + // ended. Send the policy close before the teardown closes the + // socket so the client reliably reads 4001 instead of 1006. + attachClose(client, attachCloseJobEnded, "X11 session ended") + } + }() + go func() { + defer wg.Done() + defer teardown() + // client-initiated end: teardown closes the worker conn so the + // TCP pump's blocking read unblocks. + xattachPumpWS(client, worker, done) + }() + + // The job ending is a session end regardless of the worker hop. + go func() { + ch := backend.DefaultBroker.Subscribe(4) + defer backend.DefaultBroker.Unsubscribe(ch) + for { + select { + case <-done: + return + case ev, ok := <-ch: + if !ok { + return + } + if ev.JobID == jobID && ev.Type == backend.EventJobFinished { + select { + case <-done: + default: + logger.Info("xattach: job finished, ending relayed session", + zap.Uint64("jobid", jobID)) + _ = client.WriteControl(websocket.CloseMessage, + websocket.FormatCloseMessage(attachCloseJobEnded, + "X11 session ended (job finished)"), + time.Now().Add(xattachWriteTimeout)) + } + go teardown() + return + } + } + } + }() + + // Either pump's exit tears both sockets down (teardown above), which + // unblocks the other pump's read; wg.Wait below just reaps them. + <-done + wg.Wait() +} + +// xattachPumpTCP copies worker TCP bytes into websocket binary frames. +// Returns false when the TCP side ended on its own (its peer outlived it, +// and the caller may want to send it a policy close). +func xattachPumpTCP(src net.Conn, dst *websocket.Conn, done <-chan struct{}) bool { + buf := make([]byte, 32*1024) + for { + n, err := src.Read(buf) + if n > 0 { + _ = dst.SetWriteDeadline(time.Now().Add(xattachWriteTimeout)) + if werr := dst.WriteMessage(websocket.BinaryMessage, buf[:n]); werr != nil { + return true // dst went away first; teardown unblocks us + } + } + if err != nil { + select { + case <-done: + return true + default: + return err == net.ErrClosed + } + } + } +} + +// xattachPumpWS copies websocket binary frames into the worker TCP +// connection. Text frames are not part of the X11 wire and are ignored. +func xattachPumpWS(src *websocket.Conn, dst net.Conn, done <-chan struct{}) { + for { + msgType, data, err := src.ReadMessage() + if err != nil { + return + } + if msgType != websocket.BinaryMessage { + continue + } + _ = dst.SetWriteDeadline(time.Now().Add(xattachWriteTimeout)) + if _, err := dst.Write(data); err != nil { + return + } + } +} diff --git a/src/frontend/apiserver/controllers/job_xattach_test.go b/src/frontend/apiserver/controllers/job_xattach_test.go new file mode 100644 index 00000000..c6114ec8 --- /dev/null +++ b/src/frontend/apiserver/controllers/job_xattach_test.go @@ -0,0 +1,47 @@ +package controllers + +import ( + "testing" +) + +// x11ForwardAddr turns the worker's reported session URL (x11://host:port, +// generate_url's X11 branch) into the raw TCP address the relay dials. +func TestX11ForwardAddr(t *testing.T) { + cases := []struct { + raw string + want string + }{ + {"x11://10.0.1.5:6100", "10.0.1.5:6100"}, + {"x11://worker.example:6199", "worker.example:6199"}, + {" x11://10.0.1.5:6100 ", "10.0.1.5:6100"}, + } + for _, c := range cases { + got, err := x11ForwardAddr(c.raw) + if err != nil { + t.Errorf("x11ForwardAddr(%q): unexpected error %v", c.raw, err) + continue + } + if got != c.want { + t.Errorf("x11ForwardAddr(%q): got %q, want %q", c.raw, got, c.want) + } + } +} + +// A ttyd or VNC URL routed at the xattach relay is a bug elsewhere in the +// chain (the URL decides the backend) -- reject it instead of dialing a +// websocket endpoint as raw TCP. +func TestX11ForwardAddr_NotX11(t *testing.T) { + cases := []string{ + "", " ", + "http://10.0.1.5:9000", + "10.0.1.5::5901", + "x11://", + "x11://10.0.1.5:6100/ws", + "x11://10.0.1.5:6100/path?x=1", + } + for _, raw := range cases { + if got, err := x11ForwardAddr(raw); err == nil { + t.Errorf("x11ForwardAddr(%q): got %q, want error", raw, got) + } + } +} diff --git a/src/frontend/apiserver/controllers/jobs.go b/src/frontend/apiserver/controllers/jobs.go index ba80eb0f..5ecb8f1d 100644 --- a/src/frontend/apiserver/controllers/jobs.go +++ b/src/frontend/apiserver/controllers/jobs.go @@ -2056,6 +2056,13 @@ func processJobRequest(ctx *gin.Context, req *models.JobSubmitRequest, defaultVa extendDataMap["attach"] = req.Attach } + // x11=1: an XJOB job running as X11 forwarding (xcecli -I -X). The + // worker exports DISPLAY to its forwarding port and reports a per-job + // xauth cookie instead of starting VNC. + if req.X11 != 0 { + extendDataMap["x11"] = req.X11 + } + if len(extendDataMap) > 0 { updatedExtendDataBytes, err := json.Marshal(extendDataMap) if err != nil { diff --git a/src/frontend/apiserver/controllers/jobs_handler_test.go b/src/frontend/apiserver/controllers/jobs_handler_test.go index 1e55408e..a5f3f387 100644 --- a/src/frontend/apiserver/controllers/jobs_handler_test.go +++ b/src/frontend/apiserver/controllers/jobs_handler_test.go @@ -478,6 +478,52 @@ func TestJobHandler_SubmitInteractiveAttachExdata(t *testing.T) { } } +// TestJobHandler_SubmitX11Exdata checks the submit handler packs x11 into +// the job's ExtendData for -X jobs: the key selects the X11-forwarding +// backend worker-side (DISPLAY to the forwarding port instead of VNC). +func TestJobHandler_SubmitX11Exdata(t *testing.T) { + cases := []struct { + name string + req map[string]interface{} + wantX11 bool + }{ + {"x11 job", map[string]interface{}{ + "job_name": "a", "command": "xclock", "interactive": 2, "x11": 1, + }, true}, + {"plain xjob (VNC backend)", map[string]interface{}{ + "job_name": "a", "command": "xclock", "interactive": 2, + }, false}, + {"tty job without x11", map[string]interface{}{ + "job_name": "a", "command": "sleep 1", "interactive": 1, + }, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newHandlerFixture(t) + f.seedQueue("normal") + + w := f.submit(tc.req) + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String()) + } + subs := f.submissions() + if len(subs) != 1 { + t.Fatalf("submitJob calls = %d, want 1", len(subs)) + } + var ex map[string]interface{} + if raw := subs[0].Job.ExtendData.String; raw != "" { + if err := json.Unmarshal([]byte(raw), &ex); err != nil { + t.Fatalf("ExtendData %q is not JSON: %v", raw, err) + } + } + _, hasX11 := ex["x11"] + if hasX11 != tc.wantX11 { + t.Errorf("ExtendData x11 present = %v, want %v (exdata: %s)", hasX11, tc.wantX11, subs[0].Job.ExtendData.String) + } + }) + } +} + // TestJobHandler_SmokeControlByID controls one seeded ordinary job by ID // and checks the dispatch is recorded once with its unique_job_id. func TestJobHandler_SmokeControlByID(t *testing.T) { diff --git a/src/frontend/apiserver/models/jobs.go b/src/frontend/apiserver/models/jobs.go index b84223af..12579654 100644 --- a/src/frontend/apiserver/models/jobs.go +++ b/src/frontend/apiserver/models/jobs.go @@ -580,6 +580,12 @@ type JobInfo struct { // JobAdditionalFields (the query-field whitelist) must stay in sync with // this name. InteractiveJobURL string `json:"interactive_job_url"` + // X11Cookie is the per-job MIT-MAGIC-COOKIE-1 of an -X job (XJOB with + // exdata x11=1), extracted from the same start-report exdata as + // InteractiveJobURL. The xcecli -X client registers it with the user's + // Xauthority before the app's first relayed X connection. Empty for + // every other kind of job. + X11Cookie string `json:"x11_cookie"` } type JobExecHosts struct { @@ -632,6 +638,7 @@ type JobAdditionalFields struct { Pids []int `json:"pids"` // List of process IDs associated with the job Pgids []int `json:"pgids"` // List of process group IDs associated with the job InteractiveJobURL string `json:"interactive_job_url"` + X11Cookie string `json:"x11_cookie"` // -X job's per-session auth cookie (from start exdata) // CoreLimit/MemLimit: JobInfo has no equivalent field (these are // submission-time resource limits, not job-state fields), but xcecli's // JobFieldsMap accepts them as valid -o/-l display fields (bug-1674) -- @@ -642,38 +649,43 @@ type JobAdditionalFields struct { } type JobSubmitRequest struct { - JobName string `json:"job_name,omitempty" ` - Command string `json:"command" binding:"required"` - QueueName string `json:"queue_name,omitempty"` // set to DEFAULT_QUEUE in config - ProjectName string `json:"project_name,omitempty"` // default project: "default" - UserGroup string `json:"user_group,omitempty"` - LicenseProject string `json:"license_project,omitempty"` - JobDescription string `json:"job_description,omitempty"` - Resreq string `json:"resreq,omitempty"` // default: "type == local" - Affinity string `json:"affinity,omitempty"` - Gpureq string `json:"gpureq,omitempty"` - RsvID string `json:"rsv_id,omitempty"` - SLA string `json:"sla,omitempty"` - MaxNumProcessors int `json:"maxnum_processors,omitempty"` - NumProcessors int `json:"num_processors,omitempty"` - AskedHosts string `json:"asked_hosts,omitempty"` - LaunchType string `json:"launch_type,omitempty"` - BeginTime CustomTime `json:"begin_time,omitempty"` //k change type - TermTime CustomTime `json:"term_time,omitempty"` //k change type - JobPriority int `json:"job_priority,omitempty"` - DependCond string `json:"depend_cond,omitempty"` - JobGroup string `json:"job_group,omitempty"` - HostSpec string `json:"host_spec,omitempty"` - Env []string `json:"env,omitempty"` - Tags []string `json:"tags,omitempty"` - Exclusive bool `json:"exclusive,omitempty"` - Rerunable bool `json:"rerunable,omitempty"` - Interactive int `json:"interactive,omitempty"` + JobName string `json:"job_name,omitempty" ` + Command string `json:"command" binding:"required"` + QueueName string `json:"queue_name,omitempty"` // set to DEFAULT_QUEUE in config + ProjectName string `json:"project_name,omitempty"` // default project: "default" + UserGroup string `json:"user_group,omitempty"` + LicenseProject string `json:"license_project,omitempty"` + JobDescription string `json:"job_description,omitempty"` + Resreq string `json:"resreq,omitempty"` // default: "type == local" + Affinity string `json:"affinity,omitempty"` + Gpureq string `json:"gpureq,omitempty"` + RsvID string `json:"rsv_id,omitempty"` + SLA string `json:"sla,omitempty"` + MaxNumProcessors int `json:"maxnum_processors,omitempty"` + NumProcessors int `json:"num_processors,omitempty"` + AskedHosts string `json:"asked_hosts,omitempty"` + LaunchType string `json:"launch_type,omitempty"` + BeginTime CustomTime `json:"begin_time,omitempty"` //k change type + TermTime CustomTime `json:"term_time,omitempty"` //k change type + JobPriority int `json:"job_priority,omitempty"` + DependCond string `json:"depend_cond,omitempty"` + JobGroup string `json:"job_group,omitempty"` + HostSpec string `json:"host_spec,omitempty"` + Env []string `json:"env,omitempty"` + Tags []string `json:"tags,omitempty"` + Exclusive bool `json:"exclusive,omitempty"` + Rerunable bool `json:"rerunable,omitempty"` + Interactive int `json:"interactive,omitempty"` // Attach marks an interactive TTY job as the terminal-attach kind // (xcecli -I without --browse): the session ends with the command, so // the worker may reap the ttyd when no client ever connects. 0 keeps // the web-viewed semantics. - Attach int `json:"attach,omitempty"` + Attach int `json:"attach,omitempty"` + // X11 (exdata key "x11") selects the X11-forwarding backend for an XJOB + // (interactive=2) job: the worker exports DISPLAY to its forwarding port + // instead of starting VNC, and the app's X connection is relayed to the + // user's local X server through the xattach websocket. + X11 int `json:"x11,omitempty"` WarningTimePeriod int `json:"warning_time_period,omitempty"` WarningAction string `json:"warning_action,omitempty"` DataSpecFile string `json:"data_spec_file,omitempty"` diff --git a/src/frontend/apiserver/router/routers.go b/src/frontend/apiserver/router/routers.go index 94116a08..57185056 100644 --- a/src/frontend/apiserver/router/routers.go +++ b/src/frontend/apiserver/router/routers.go @@ -147,6 +147,9 @@ func Router() *gin.Engine { // Interactive-job websocket relay (attach): long-lived connection to // the job's ttyd session on the worker, proxied by the apiserver. jobsApi.GET("/:jobids/attach", controllers.RequirePermission("job:control"), jobHandler.AttachJob) + // X11-forwarding relay (-X jobs): one websocket per X connection of + // the job's forwarding port, raw TCP on the worker side. + jobsApi.GET("/:jobids/xattach", controllers.RequirePermission("job:control"), jobHandler.XAttachJob) jobsApi.GET("", controllers.RequirePermission("job:view"), jobHandler.GetJobsInfo) jobsApi.GET("/:jobids", controllers.RequirePermission("job:view"), jobHandler.GetJobByIDs) jobsApi.POST("/create", controllers.RequirePermission("queue:submit"), jobHandler.CreateJob) diff --git a/src/frontend/xcecli/cmd/jobs/submit.go b/src/frontend/xcecli/cmd/jobs/submit.go index ce9f2178..8250b0ca 100755 --- a/src/frontend/xcecli/cmd/jobs/submit.go +++ b/src/frontend/xcecli/cmd/jobs/submit.go @@ -57,6 +57,7 @@ type JobSubmitter struct { interactive bool useTTY bool useXJob bool + useX11 bool useBrowse bool reportedStart bool projectName string @@ -89,16 +90,8 @@ func (js *JobSubmitter) ParseCmdArgs(cmd *cobra.Command, args []string) error { } if js.interactive { - if js.useTTY && js.useXJob { - return fmt.Errorf("error: cannot use both --tty and --xjob") - } - // Bare -I is the attach interactive session: -I without --xjob is - // equivalent to -I --tty. The old mapping (interactive=3, readonly - // DISPLAY web view) had no interactive input and no user hint; the - // API's interactive=3 semantics are unchanged -- only this CLI's - // default remaps to TTY attach (use --xjob for VNC). - if !js.useXJob { - js.useTTY = true + if err := js.applyInteractiveDefaults(); err != nil { + return err } } @@ -243,6 +236,31 @@ func (js *JobSubmitter) parseConfigFile() error { } +// applyInteractiveDefaults resolves the -I flag family into the concrete +// interactive kind. Bare -I is the attach interactive session: -I without +// --xjob is equivalent to -I --tty. The old mapping (interactive=3, readonly +// DISPLAY web view) had no interactive input and no user hint; the API's +// interactive=3 semantics are unchanged -- only this CLI's default remaps to +// TTY attach (use --xjob for VNC). -I -X runs a graphical X11 app in a popup +// window on this display (ssh -X experience): interactive=2 (XJOB) with the +// x11 backend selected worker-side by exdata; it suppresses the TTY remap. +func (js *JobSubmitter) applyInteractiveDefaults() error { + if js.useTTY && js.useXJob { + return fmt.Errorf("error: cannot use both --tty and --xjob") + } + if js.useX11 && js.useBrowse { + return fmt.Errorf("error: cannot use both --x11 and --browse") + } + if !js.useXJob { + js.useTTY = true + } + if js.useX11 { + js.useTTY = false + js.useXJob = true + } + return nil +} + // BuildJobRequest func (js *JobSubmitter) BuildJobRequest() (*models.JobSubmitRequest, error) { // create the job request with only required fields @@ -261,6 +279,12 @@ func (js *JobSubmitter) BuildJobRequest() (*models.JobSubmitRequest, error) { } } else if js.useXJob { jobReq.Interactive = 2 // XJob mode + // -X: the X11-forwarding backend (popup window on the user's own + // display) instead of VNC; the worker picks the backend from this + // exdata key, so the interactive enum stays untouched. + if js.useX11 { + jobReq.X11 = 1 + } } else { jobReq.Interactive = 3 // only interactive, no special mode } @@ -540,6 +564,14 @@ func (js *JobSubmitter) RunJobsSubmit() error { return nil } + // -X (X11 forwarding): the app's X connection is relayed through the + // apiserver to this display -- windows pop up here like ssh -X. The + // URL is a worker-internal detail; the session ends with the command + // and the CLI pumps the relay until the job finishes. + if js.useX11 { + return runX11Session(serverInfo.URL, serverInfo.Token, jobID) + } + // XJOB (VNC) has no terminal rendering -- browser remains the path fmt.Printf("interactive session URL: %s\n", url) if err := openBrowser(url); err != nil { @@ -765,6 +797,7 @@ Example: xce jobs submit -I --tty sleep 10 # submit interactive job with tty and attach in this terminal xce jobs submit -I --tty --browse sleep 10 # submit interactive tty job, open the web URL in a browser xce jobs submit -I --xjob sleep 10 # submit interactive job with xjob (VNC) + xce jobs submit -I -X xclock # run xclock on the cluster, window pops up on this display (ssh -X style) xce jobs submit -J 'name[1-N]' sleep 10 # submit a job array of N elements named "name" xce jobs submit -J 'name[1]' sleep 10 # submit a job array of one element @@ -829,9 +862,10 @@ Job arrays: flags.StringVarP(&jobSubmitter.outdir, "outdir", "d", "", "specify the output directory") flags.StringVarP(&jobSubmitter.projectName, "project_name", "P", "", "specify the project name") flags.StringVar(&jobSubmitter.configFile, "configfile", "", "specify the configuration file for job submission") - flags.BoolVarP(&jobSubmitter.interactive, "interactive", "I", false, "submit interactive job (attach in this terminal; same as -I --tty; add --xjob for VNC)") + flags.BoolVarP(&jobSubmitter.interactive, "interactive", "I", false, "submit interactive job (attach in this terminal; same as -I --tty; add --xjob for VNC; add -X for X11 popup)") flags.BoolVar(&jobSubmitter.useTTY, "tty", false, "use TTY for interactive job (attach in this terminal after dispatch; --browse opens the web URL instead; default for -I)") flags.BoolVar(&jobSubmitter.useXJob, "xjob", false, "use XJob (VNC) for interactive job") + flags.BoolVarP(&jobSubmitter.useX11, "x11", "X", false, "run a graphical X11 app and show its windows on this display (ssh -X style; requires a local X server and xauth)") flags.BoolVar(&jobSubmitter.useBrowse, "browse", false, "open the interactive session URL in a browser instead of attaching (TTY jobs attach by default)") flags.StringVarP(&jobSubmitter.askedHosts, "m", "m", "", "specify the running hosts, with space for multiple values") flags.StringVarP(&jobSubmitter.launchType, "launch_type", "", "", "specify the launch type") diff --git a/src/frontend/xcecli/cmd/jobs/x11.go b/src/frontend/xcecli/cmd/jobs/x11.go new file mode 100644 index 00000000..68efd0c9 --- /dev/null +++ b/src/frontend/xcecli/cmd/jobs/x11.go @@ -0,0 +1,311 @@ +package jobs + +import ( + "fmt" + "net" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + "github.com/gorilla/websocket" + + "xcecli/common/request" + "xcecli/models" +) + +// X11-forwarding session for an interactive XJOB job (-I -X). +// +// The worker exported DISPLAY to its forwarding port and starts the app +// immediately (no daemon waits for a client). The app connects to that port; +// the apiserver's /xattach relay bridges each worker-side TCP connection to +// one websocket; this side pumps each websocket into the user's local X +// server, so the app's windows appear on the user's display exactly like +// ssh -X. Rendering happens client-side: no VNC server, no browser. +// +// One websocket = one X connection = one worker-side accept. An X app opens +// one connection on XOpenDisplay (secondary ones, e.g. xim, are rare and not +// supported in v1 -- the worker refuses them loudly rather than dropping +// events silently). +// +// Auth is ssh -X's model: the worker generates a per-job MIT-MAGIC-COOKIE-1 +// and reports it in the job's exdata; we add it to the user's Xauthority +// before the app's first connection can arrive and remove it when the +// session ends. The job detail API carries the cookie next to the URL. + +// x11DisplayAddr resolves $DISPLAY into a dialable address of the local X +// server: a unix socket path or a TCP host:port. Only local displays are +// supported -- the whole point of -X is popping windows up here. +func x11DisplayAddr(display string) (string, bool, error) { + display = strings.TrimSpace(display) + if display == "" { + return "", false, fmt.Errorf("DISPLAY is not set: -X shows windows on this machine's X server, run from a graphical session (or inside ssh -X)") + } + // X syntax: [host]:N[.S]; an empty host means the local unix socket. + idx := strings.LastIndex(display, ":") + if idx < 0 { + return "", false, fmt.Errorf("cannot parse DISPLAY %q (expected [host]:N)", display) + } + host := display[:idx] + num := display[idx+1:] + if dot := strings.Index(num, "."); dot >= 0 { + num = num[:dot] + } + n := 0 + if _, err := fmt.Sscanf(num, "%d", &n); err != nil { + return "", false, fmt.Errorf("cannot parse display number in DISPLAY %q", display) + } + if host == "" || host == "unix" { + return fmt.Sprintf("/tmp/.X11-unix/X%d", n), true, nil + } + // A remote-looking host is only valid when it is us (X server bound to + // an interface) -- TCP to host:6000+N either way; ssh -X users normally + // have "localhost:10.0" (their own ssh-forwarded display), which is a + // local TCP listener owned by sshd and works the same. + return fmt.Sprintf("%s:%d", host, 6000+n), false, nil +} + +// x11XauthFile picks the Xauthority file xauth should touch: $XAUTHORITY if +// set, else ~/.Xauthority. ssh -X sessions set XAUTHORITY to a per-session +// file, which is exactly the one the local X server honors. +func x11XauthFile() (string, error) { + if f := os.Getenv("XAUTHORITY"); f != "" { + return f, nil + } + home, err := os.UserHomeDir() + if err != nil { + return "", fmt.Errorf("cannot locate the Xauthority file (no XAUTHORITY, no home): %s", err) + } + return filepath.Join(home, ".Xauthority"), nil +} + +// x11Xauth runs xauth with the given arguments, scoped to our auth file. +func x11Xauth(authFile string, args ...string) error { + cmd := exec.Command("xauth", append([]string{"-f", authFile}, args...)...) + if out, err := cmd.CombinedOutput(); err != nil { + return fmt.Errorf("xauth %s failed: %s (%s)", strings.Join(args, " "), + attachErrText(err), strings.TrimSpace(string(out))) + } + return nil +} + +// x11Cookie fetches the job's X11 auth cookie reported by the worker +// (exdata "x11cookie") alongside the session URL. +func x11Cookie(serverURL, token, jobID string) (string, error) { + srv, err := x11ServerBase(serverURL) + if err != nil { + return "", err + } + var resp models.JobsResponse + client := request.NewAPIClient(srv) + err = client.GetRequestFromAPIServer( + "Bearer "+token, + "", + fmt.Sprintf("/xce/v1/jobs/%s", jobID), + &resp, + map[string]string{"fields": "x11_cookie"}, + ) + if err != nil { + return "", fmt.Errorf("failed to get job details: [%s]", request.ExtractErrorMessage(err)) + } + for _, row := range resp.Data.Rows { + if row.X11Cookie != "" { + return row.X11Cookie, nil + } + } + return "", fmt.Errorf("job <%s> has no X11 session cookie (not an -X job, not running yet, or an older worker without X11 support)", jobID) +} + +// x11ServerBase normalizes a configured server URL to the http(s) API base +// (attachEndpoint's host extraction, minus the websocket scheme swap). +func x11ServerBase(serverURL string) (string, error) { + base := strings.TrimRight(strings.TrimSpace(serverURL), "/") + if base == "" { + return "", fmt.Errorf("no API server URL configured") + } + if idx := strings.Index(base, "/xce/"); idx != -1 { + base = base[:idx] + } + if !strings.HasPrefix(base, "http://") && !strings.HasPrefix(base, "https://") { + return "", fmt.Errorf("unsupported API server URL scheme: %s", serverURL) + } + return base, nil +} + +// runX11Session wires the relayed X connection(s) of one job into the local +// X server until the session (its job) ends, then cleans the cookie up. +func runX11Session(serverURL, token, jobID string) error { + cookie, err := x11Cookie(serverURL, token, jobID) + if err != nil { + return err + } + + display := os.Getenv("DISPLAY") + addr, isUnix, err := x11DisplayAddr(display) + if err != nil { + return err + } + authFile, err := x11XauthFile() + if err != nil { + return err + } + if _, err := exec.LookPath("xauth"); err != nil { + return fmt.Errorf("xauth not found in PATH: X11 forwarding needs it to register the session's auth cookie (an X server is present, as DISPLAY=%s suggests)", display) + } + + // The worker's X server-side auth check happens against the cookie we + // are about to install; the app may connect the moment we open the + // relay, so the cookie must be in place first. xauth takes the display + // name, auth name and cookie as three separate arguments -- one combined + // argument is a "bad add command line". + if err := x11Xauth(authFile, "add", display, ".", cookie); err != nil { + return err + } + removed := false + removeCookie := func() { + if removed { + return + } + removed = true + // Best-effort: a leftover entry expires only in that it stops + // matching anything once the job's display is gone. + _ = x11Xauth(authFile, "remove", display) + } + defer removeCookie() + + fmt.Printf("attaching X11 session of job <%s> to display %s ...\n", jobID, display) + endpoint, err := attachEndpoint(serverURL, jobID) + endpoint = strings.Replace(endpoint, "/attach", "/xattach", 1) + if err != nil { + return err + } + + sess := &x11Session{ + endpoint: endpoint, + token: token, + addr: addr, + isUnix: isUnix, + } + err = sess.run() + + // The job ending is the normal end of an -X session (the command's exit + // tears the relay down); a connection error before any data flowed is + // the interesting failure -- report it. + if err != nil { + if ce, ok := err.(*websocket.CloseError); ok && (ce.Code == 4001 || ce.Code == websocket.CloseNormalClosure) { + fmt.Println("<>") + return nil + } + return err + } + fmt.Println("<>") + return nil +} + +type x11Session struct { + endpoint string + token string + addr string + isUnix bool +} + +// run dials the relay once and pumps it into the local X server. The first +// worker-side accept (the app's XOpenDisplay) rides this connection. +func (s *x11Session) run() error { + xc, err := x11DialLocal(s.addr, s.isUnix) + if err != nil { + return fmt.Errorf("cannot reach the local X server at %s: %s (is the display still alive?)", s.addr, attachErrText(err)) + } + defer xc.Close() + + ws, _, err := websocket.DefaultDialer.Dial(s.endpoint, + map[string][]string{"Authorization": {"Bearer " + s.token}}) + if err != nil { + return fmt.Errorf("xattach failed: [%s]", attachErrText(err)) + } + defer ws.Close() + + done := make(chan struct{}) + errCh := make(chan error, 2) + + // local X server -> relay: the app talks to its (remote) display here + go func() { + buf := make([]byte, 32*1024) + for { + n, err := xc.Read(buf) + if n > 0 { + if werr := ws.WriteMessage(websocket.BinaryMessage, buf[:n]); werr != nil { + errCh <- werr + return + } + } + if err != nil { + if n == 0 { + errCh <- err // X connection closed: the app is done with it + } + return + } + } + }() + + // relay -> local X server + go func() { + for { + mt, data, err := ws.ReadMessage() + if err != nil { + select { + case <-done: + default: + errCh <- err + } + return + } + if mt != websocket.BinaryMessage { + continue + } + if _, err := xc.Write(data); err != nil { + errCh <- err + return + } + } + }() + + // Either side ending tears both down; report the first real error (the + // websocket close usually follows the app exiting, so a bare 1006 with + // the X side already gone is a normal end). + select { + case err := <-errCh: + close(done) + if isX11NormalEnd(err) { + return nil + } + return err + case <-done: + return nil + } +} + +// isX11NormalEnd reports whether err is the session ending the way an +// ssh -X session does: the app exited (X connection EOF -> relay torn down +// -> abrupt websocket close) or the job finished (policy close 4001). +func isX11NormalEnd(err error) bool { + if err == nil { + return true + } + if websocket.IsCloseError(err, 4001, websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { + return true + } + // A bare "unexpected EOF"/io.ErrUnexpectedEOF from a torn-down TCP leg + // is the app-exit race, same as attach's 1006-after-output handling. + return err == net.ErrClosed || strings.Contains(err.Error(), "unexpected EOF") +} + +// x11DialLocal opens one connection to the local X server (unix socket or +// TCP depending on how DISPLAY points at it). +func x11DialLocal(addr string, isUnix bool) (net.Conn, error) { + if isUnix { + return net.DialTimeout("unix", addr, 5*time.Second) + } + return net.DialTimeout("tcp", addr, 5*time.Second) +} diff --git a/src/frontend/xcecli/cmd/jobs/x11_test.go b/src/frontend/xcecli/cmd/jobs/x11_test.go new file mode 100644 index 00000000..a80157ee --- /dev/null +++ b/src/frontend/xcecli/cmd/jobs/x11_test.go @@ -0,0 +1,182 @@ +package jobs + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "xcecli/models" +) + +// -X maps to interactive=2 (XJOB) with the x11 exdata key selecting the +// forwarding backend worker-side; the interactive enum itself stays +// untouched (master/worker ABI). The matrix runs the same interactive +// defaults the cobra flow applies before BuildJobRequest. +func TestBuildJobRequest_X11FlagMatrix(t *testing.T) { + cases := []struct { + name string + interactive bool + useTTY bool + useX11 bool + useXJob bool + useBrowse bool + wantInter int + wantX11 int + }{ + {"x11 job (-I -X)", true, false, true, false, false, 2, 1}, + {"bare -I unchanged", true, false, false, false, false, 1, 0}, + {"tty unchanged", true, true, false, false, false, 1, 0}, + {"xjob unchanged", true, false, false, true, false, 2, 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + js := NewJobSubmitter(nil) + js.command = "xclock" + js.interactive = tc.interactive + js.useTTY = tc.useTTY + js.useX11 = tc.useX11 + js.useXJob = tc.useXJob + js.useBrowse = tc.useBrowse + if tc.interactive { + if err := js.applyInteractiveDefaults(); err != nil { + t.Fatalf("applyInteractiveDefaults: %v", err) + } + } + req, err := js.BuildJobRequest() + if err != nil { + t.Fatalf("BuildJobRequest: %v", err) + } + if req.Interactive != tc.wantInter { + t.Errorf("Interactive = %d, want %d", req.Interactive, tc.wantInter) + } + if req.X11 != tc.wantX11 { + t.Errorf("X11 = %d, want %d", req.X11, tc.wantX11) + } + }) + } +} + +// -I -X conflicts with the TTY remap and with --browse: the interactive +// defaults must reject the combinations before anything is submitted. +func TestInteractiveDefaults_X11Conflicts(t *testing.T) { + js := NewJobSubmitter(nil) + js.interactive = true + js.useX11 = true + js.useBrowse = true + if err := js.applyInteractiveDefaults(); err == nil { + t.Fatal("--x11 with --browse should be rejected") + } +} + +func TestInteractiveDefaults_X11Remap(t *testing.T) { + js := NewJobSubmitter(nil) + js.interactive = true + js.useX11 = true + if err := js.applyInteractiveDefaults(); err != nil { + t.Fatalf("applyInteractiveDefaults: %v", err) + } + if js.useTTY { + t.Error("useTTY should be cleared for -X (not a terminal session)") + } + if !js.useXJob { + t.Error("useXJob should be set for -X (interactive=2)") + } +} + +// Bare -I keeps the TTY attach remap when -X is absent. +func TestInteractiveDefaults_BareTTYUnchanged(t *testing.T) { + js := NewJobSubmitter(nil) + js.interactive = true + if err := js.applyInteractiveDefaults(); err != nil { + t.Fatalf("applyInteractiveDefaults: %v", err) + } + if !js.useTTY || js.useXJob { + t.Errorf("bare -I = (tty=%v, xjob=%v), want (true, false)", js.useTTY, js.useXJob) + } +} + +// DISPLAY parsing: local unix socket forms, TCP forms, and the errors that +// keep -X from silently pointing at something that is not this display. +func TestX11DisplayAddr(t *testing.T) { + cases := []struct { + display string + wantAddr string + wantUnix bool + wantErr bool + }{ + {":0", "/tmp/.X11-unix/X0", true, false}, + {":12.0", "/tmp/.X11-unix/X12", true, false}, + {"unix:0", "/tmp/.X11-unix/X0", true, false}, + {"localhost:0", "localhost:6000", false, false}, + {"localhost:10.0", "localhost:6010", false, false}, // ssh -X forwarded display + {"", "", false, true}, + {"bogus", "", false, true}, + {":", "", false, true}, + {":x", "", false, true}, + } + for _, tc := range cases { + t.Run(tc.display, func(t *testing.T) { + addr, isUnix, err := x11DisplayAddr(tc.display) + if tc.wantErr { + if err == nil { + t.Fatalf("DISPLAY %q: want error, got %q", tc.display, addr) + } + return + } + if err != nil { + t.Fatalf("DISPLAY %q: %v", tc.display, err) + } + if addr != tc.wantAddr || isUnix != tc.wantUnix { + t.Errorf("DISPLAY %q = (%q, %v), want (%q, %v)", tc.display, addr, isUnix, tc.wantAddr, tc.wantUnix) + } + }) + } +} + +// The cookie fetch reads the job detail's x11_cookie field (the apiserver +// extracts it from the worker's start-report exdata). +func TestX11Cookie_FromJobDetail(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("fields") != "x11_cookie" { + t.Errorf("fields = %q, want x11_cookie", r.URL.Query().Get("fields")) + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(models.JobsResponse{ + Code: 0, + Data: struct { + Count int `json:"count"` + Rows []models.QueryJob `json:"rows"` + Page_token string `json:"page_token"` + }{Rows: []models.QueryJob{{JobStatus: "RUN", X11Cookie: "aabbccdd00112233"}}}, + }) + })) + t.Cleanup(srv.Close) + cookie, err := x11Cookie(srv.URL, "tok", "123") + if err != nil { + t.Fatalf("x11Cookie: %v", err) + } + if cookie != "aabbccdd00112233" { + t.Errorf("cookie = %q", cookie) + } +} + +// A job without a cookie (wrong kind, older worker) must fail with a hint, +// not with an empty cookie that xauth would choke on. +func TestX11Cookie_Missing(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(models.JobsResponse{ + Code: 0, + Data: struct { + Count int `json:"count"` + Rows []models.QueryJob `json:"rows"` + Page_token string `json:"page_token"` + }{Rows: []models.QueryJob{{JobStatus: "RUN"}}}, + }) + })) + t.Cleanup(srv.Close) + if _, err := x11Cookie(srv.URL, "tok", "123"); err == nil { + t.Fatal("want error for a job without x11_cookie") + } +} diff --git a/src/frontend/xcecli/models/jobs.go b/src/frontend/xcecli/models/jobs.go index 53dfb43d..46f3a2dd 100755 --- a/src/frontend/xcecli/models/jobs.go +++ b/src/frontend/xcecli/models/jobs.go @@ -191,6 +191,9 @@ type QueryJob struct { UniqueJobID int64 `json:"unique_job_id"` // ExtendData JobExtendData `json:"extend_data"` InteractiveJobURL string `json:"interactive_job_url"` + // X11Cookie carries the per-job MIT-MAGIC-COOKIE-1 of an -X job, + // extracted from the worker's start-report exdata by the apiserver. + X11Cookie string `json:"x11_cookie"` PGIDs IntSlice `json:"pgids"` PID IntSlice `json:"pids"` LaunchType string `json:"launch_type"` @@ -348,6 +351,11 @@ type JobSubmitRequest struct { // may reap the ttyd when no client ever connects. 0 (default) keeps the // web-viewed semantics. Attach int `json:"attach,omitempty"` + // X11 (exdata key "x11") selects the X11-forwarding backend for an XJOB + // (interactive=2) job: the worker exports DISPLAY to its forwarding port + // instead of starting VNC, and the app's X connection is relayed to the + // user's local X server through the apiserver xattach websocket. + X11 int `json:"x11,omitempty"` WarningTimePeriod int `json:"warning_time_period,omitempty"` WarningAction string `json:"warning_action,omitempty"` DataSpecFile string `json:"data_spec_file,omitempty"` diff --git a/src/worker/include/inter_job_utils.h b/src/worker/include/inter_job_utils.h index 4f2aaf04..dbdf7533 100644 --- a/src/worker/include/inter_job_utils.h +++ b/src/worker/include/inter_job_utils.h @@ -5,6 +5,11 @@ #define URL_MAX_LENGTH 256 #define TTY_END_PORT 10000 #define XJOB_END_PORT 5999 +/* X11 forwarding (-X jobs, exdata "x11":1): the worker listens on this + * segment and exports DISPLAY=: to the job; the app's X + * connections are relayed to the user's X server through the apiserver. */ +#define X11_START_PORT 6100 +#define X11_END_PORT 6199 typedef struct { int tty_port; @@ -17,5 +22,6 @@ int check_port(port_info_t port, int interactive_mode); char* generate_cmd(char* job_cmd, port_info_t port, int interactive_mode); char* generate_url(port_info_t port, int interactive_mode); int export_display_env(int port); +int export_x11_display_env(int port); #endif /* _INTER_JOB_UTILS_H */ diff --git a/src/worker/include/xcecd_exdata.h b/src/worker/include/xcecd_exdata.h index 8c9a9f5c..05193881 100644 --- a/src/worker/include/xcecd_exdata.h +++ b/src/worker/include/xcecd_exdata.h @@ -28,11 +28,19 @@ typedef struct { * command (xcecli attach path). The worker may reap the ttyd when no * client ever connects; web-viewed jobs (0) keep it for a later browser. */ int attach_mode; + /* x11=1 (exdata key "x11"): XJOB job running as X11 forwarding -- the + * worker exports DISPLAY to its forwarding port and reports a per-job + * xauth cookie instead of starting VNC (xcecli -I -X). */ + int x11_mode; int count; } transfer_info_t; typedef struct { char* url; + /* Per-job MIT-MAGIC-COOKIE-1 (hex) for X11-forwarding jobs, generated + * with xauth and reported to the apiserver so the user's client can + * register it with their Xauthority. NULL otherwise. */ + char* x11cookie; } start_exdata_t; bool parse_exdata(const char* exdata, env_var_t** env_vars, diff --git a/src/worker/src/inter_job_utils.c b/src/worker/src/inter_job_utils.c index de78fa87..c75125d7 100644 --- a/src/worker/src/inter_job_utils.c +++ b/src/worker/src/inter_job_utils.c @@ -326,3 +326,35 @@ int export_display_env(int port) "Exported DISPLAY environment variable: %s", display_env); return 0; } + +/** + * @brief Export DISPLAY for an X11-forwarding job: the app's X connections + * go to the worker's forwarding port (rendering happens on the user's + * own X server via the apiserver relay, ssh -X style). + * @param port The forwarding port number (X11_START_PORT..X11_END_PORT). + * @return 0 on success, -1 on failure. + */ +int export_x11_display_env(int port) +{ + if (port < 6000 || port - 6000 > 255) { + XCECD_WORKER_LOG(XCE_LOG_ERROR, + "Invalid X11 forwarding port %d for DISPLAY", port); + return -1; + } + + char display_env[32]; + snprintf(display_env, sizeof(display_env), ":%d", port - 6000); + + if (setenv("DISPLAY", display_env, 1) != 0) { + XCECD_WORKER_LOG(XCE_LOG_ERROR, + "Failed to set DISPLAY environment variable for " + "X11 forwarding"); + return -1; + } + + XCECD_WORKER_LOG(XCE_LOG_DEBUG3, + "Exported DISPLAY environment variable for X11 " + "forwarding: %s", + display_env); + return 0; +} diff --git a/src/worker/src/xcecd_exdata.c b/src/worker/src/xcecd_exdata.c index 8e559a28..78e6163c 100644 --- a/src/worker/src/xcecd_exdata.c +++ b/src/worker/src/xcecd_exdata.c @@ -252,6 +252,22 @@ bool parse_exdata(const char* exdata, env_var_t** env_vars, XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed attach mode: %d", transfer_infos->attach_mode); } + // --- Handle 'x11' --- + else if (strcmp(key, "x11") == 0) { + if (cJSON_IsString(item) && item->valuestring != NULL) { + transfer_infos->x11_mode = atoi(item->valuestring); + } else if (cJSON_IsNumber(item)) { + transfer_infos->x11_mode = item->valueint; + } else { + XCECD_WORKER_LOG( + XCE_LOG_ERROR, + "Error: 'x11' is neither a string nor a number."); + success = false; + continue; + } + XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed x11 mode: %d", + transfer_infos->x11_mode); + } // --- Skip 'limits' --- else if (strcmp(key, "limits") == 0) { // limits will be handled in parse_limit @@ -525,6 +541,9 @@ char* generate_exdata(const start_exdata_t* start_info) } cJSON_AddStringToObject(root, "url", start_info->url); + if (start_info->x11cookie) { + cJSON_AddStringToObject(root, "x11cookie", start_info->x11cookie); + } char* exdata_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); diff --git a/src/worker/src/xcecd_worker.c b/src/worker/src/xcecd_worker.c index e5fe4982..8dafcd0a 100644 --- a/src/worker/src/xcecd_worker.c +++ b/src/worker/src/xcecd_worker.c @@ -1,4 +1,5 @@ #include "job_signal.h" +#include "job_output_path.h" #include "inter_job_utils.h" #include "process_manager.h" #include "static_array.h" @@ -8,12 +9,18 @@ #include "xcecd_exdata.h" #include "xcecd_utils.h" #include "xcelog.h" +#include #include +#include +#include +#include +#include #include #include #include #include #include +#include #include #include #include @@ -26,6 +33,12 @@ #define LOG_SIZE (1024 * 1024 * 100) // 100 MB #define LOG_LEVEL 3 +/* X11 forwarding (-X jobs): the worker listens for the app's X connections + * and exports DISPLAY to the listening port; rendering happens on the + * user's own X server via the apiserver xattach relay (ssh -X model). One + * relayed websocket carries one X connection. */ +#define X11_NO_CLIENT_MS 60000 /* reap when no X connection ever arrived */ + typedef struct exdata { env_var_t* env_vars; transfer_info_t* transfer_info; @@ -69,6 +82,37 @@ static bool tty_attach_saw_child = false; static int64_t tty_attach_child_gone_ms = 0; static const char* tty_reap_reason = NULL; +/* One end of a bridged X11 session: either the job app's X connection or + * the apiserver xattach relay connection. The two are told apart by source + * address, not arrival order: the app always runs on this host and its + * DISPLAY is always ":N", so app connections come from loopback, while the + * relay dials the reported x11://: URL from the apiserver + * host (which is never loopback, even when apiserver and worker share a + * host). Byte pumps are identical for both roles. */ +#define X11_STASH_SIZE 32768 +typedef struct { + int fd; /* connection fd, -1 = open slot */ + bool relay; /* true = apiserver relay leg */ + int len; /* bytes stashed awaiting the peer */ + char stash[X11_STASH_SIZE]; +} x11_leg_t; + +/* X11 forwarding session state (one per-job worker process, so a single + * static session is enough -- same shape as the VNC branch). */ +typedef struct { + int listen_fd; /* forwarding-port listener, -1 = off */ + int x11_port; /* TCP port DISPLAY points at */ + volatile sig_atomic_t accepts; /* total accepted X connections */ + bool saw_client; /* app+relay bridge ever established */ + uv_timer_t no_client_timer; + bool timer_ready; + char display[32]; /* ":N" exported to the job */ + char cookie_hex[65]; /* per-job MIT-MAGIC-COOKIE-1 */ + char auth_file[256]; /* per-job Xauthority file, "" = none */ + x11_leg_t legs[2]; /* the bridged pair */ +} x11_forward_t; +static x11_forward_t x11_fwd = { .listen_fd = -1 }; + // process manager will share the same event loop with ipc uv_loop_t* uv_loop = NULL; @@ -77,6 +121,12 @@ static void worker_reconnect_callback(bool connected); static int xjob_start_vnc(); static void xjob_stop_vnc(); static void tty_stop_attach_session(); +static int x11_start_forwarding(); +static void x11_stop_forwarding(); +static void x11_accept_and_pump(); +static void x11_no_client_callback(uv_timer_t* timer); +static void x11_no_client_timer_start(); +static void x11_no_client_timer_stop(); static bool check_process_status(pid_t target_pid, int expected_signal, int timeout_seconds); static pid_t ttyd_real_pid(pid_t monitored_pid); @@ -537,7 +587,12 @@ static void process_exit_callback(int pid, int exit_status, int term_signal, /* Job post-exec start here (blocked) */ if (XJOB == mydata.transfer_info->interactive_mode) { - xjob_stop_vnc(); + if (1 == mydata.transfer_info->x11_mode) { + x11_no_client_timer_stop(); + x11_stop_forwarding(); + } else { + xjob_stop_vnc(); + } } /* attach-mode TTY job: the session ends with the command (web-viewed @@ -630,6 +685,450 @@ static void check_and_reconnect() reconnect_retry_limit = 5; } +/* ---- X11 forwarding (XJOB + exdata x11=1, the xcecli -I -X path) ---- + * + * The job command starts immediately (no daemon waits for a client) with + * DISPLAY pointing at the forwarding port. The app's XOpenDisplay connects + * to the listener; the apiserver's /xattach relay picks each accepted + * connection up as one websocket and pumps the bytes to the user's X + * server, where the windows render. Auth follows ssh -X: a per-job + * MIT-MAGIC-COOKIE-1 is generated with xauth (the only X-aware helper + * guaranteed around an X11 user setup) and reported through the start + * exdata so the user's client can register it with their Xauthority. + * + * The no-client timer covers the one residue shape an -X job has: a job + * whose X connection never arrives (the CLI died before relaying) would + * otherwise sit in RUN forever -- the app blocks inside XOpenDisplay and + * never exits on its own. Once any connection was accepted the session is + * the app's and X's to end (a dropped relay surfaces as an X IO error and + * the app exits), so the timer is cancelled on the first accept. */ + +/* 32 hex chars = 128 bits of cookie, exactly xauth's MIT-MAGIC-COOKIE-1 + * format. Reads /dev/urandom; retries a few times before giving up. */ +static bool x11_generate_cookie(char* out, size_t outlen) +{ + unsigned char raw[16]; + static const char hex[] = "0123456789abcdef"; + + if (outlen < sizeof(raw) * 2 + 1) { + return false; + } + + int fd = open("/dev/urandom", O_RDONLY); + if (fd < 0) { + return false; + } + size_t got = 0; + bool ok = true; + while (got < sizeof(raw)) { + ssize_t n = read(fd, (char*)raw + got, sizeof(raw) - got); + if (n <= 0) { + if (n < 0 && errno == EINTR) { + continue; + } + ok = false; + break; + } + got += (size_t)n; + } + close(fd); + if (!ok) { + return false; + } + + for (size_t i = 0; i < sizeof(raw); i++) { + out[i * 2] = hex[raw[i] >> 4]; + out[i * 2 + 1] = hex[raw[i] & 0xF]; + } + out[sizeof(raw) * 2] = '\0'; + return true; +} + +/* Create a fresh Xauthority file for this job and register the generated + * cookie in it under our display number. The per-job worker runs on a headless + * node -- there is no session Xauthority to inherit (getenv("XAUTHORITY") is + * normally empty here), so the file is ours from scratch, named after the job + * output dir convention in /tmp. The job processes inherit the worker's + * environment, so exporting XAUTHORITY alongside DISPLAY (the same channel + * export_x11_display_env uses) makes the app's XOpenDisplay present exactly + * this cookie. */ +static int x11_install_cookie(uint64_t jobid) +{ + char cmd[MAXLINELEN]; + char jobidStr[64]; + fmt_jobid(jobidStr, sizeof(jobidStr), jobid); + + snprintf(x11_fwd.auth_file, sizeof(x11_fwd.auth_file), + XCECD_JOB_OUTPUT_DIR "/job_%s_xauth", jobidStr); + unlink(x11_fwd.auth_file); /* a re-run job must not inherit an old entry */ + + snprintf(cmd, sizeof(cmd), + "xauth -f '%s' add ':%d' . %s 2>/dev/null && chmod 600 '%s'", + x11_fwd.auth_file, x11_fwd.x11_port - 6000, x11_fwd.cookie_hex, + x11_fwd.auth_file); + /* xce-fork-ok: xcecd_worker is single-threaded (tech.md) */ + int rc = system(cmd); + if (rc != 0 || access(x11_fwd.auth_file, R_OK) != 0) { + XCECD_WORKER_LOG(XCE_LOG_ERROR, + "xauth add failed for X11 forwarding display :%d " + "(rc=%d)", + x11_fwd.x11_port - 6000, rc); + unlink(x11_fwd.auth_file); + x11_fwd.auth_file[0] = '\0'; + return -1; + } + + if (setenv("XAUTHORITY", x11_fwd.auth_file, 1) != 0) { + XCECD_WORKER_LOG(XCE_LOG_ERROR, + "Failed to set XAUTHORITY for X11 forwarding: %s", + strerror(errno)); + unlink(x11_fwd.auth_file); + x11_fwd.auth_file[0] = '\0'; + return -1; + } + return 0; +} + +static void x11_remove_cookie() +{ + if (x11_fwd.auth_file[0] != '\0') { + unlink(x11_fwd.auth_file); + x11_fwd.auth_file[0] = '\0'; + } + x11_fwd.cookie_hex[0] = '\0'; +} + +static int x11_start_forwarding() +{ + char jobidStr[64]; + fmt_jobid(jobidStr, sizeof(jobidStr), jobid); + + x11_fwd.x11_port = port_info.xjob_port; /* X11 rides the XJOB slot */ + x11_fwd.listen_fd = -1; + x11_fwd.accepts = 0; + x11_fwd.saw_client = false; + x11_fwd.cookie_hex[0] = '\0'; + x11_fwd.auth_file[0] = '\0'; + x11_fwd.legs[0].fd = -1; + x11_fwd.legs[1].fd = -1; + x11_fwd.legs[0].relay = false; + x11_fwd.legs[1].relay = false; + x11_fwd.legs[0].len = 0; + x11_fwd.legs[1].len = 0; + + if (!x11_generate_cookie(x11_fwd.cookie_hex, + sizeof(x11_fwd.cookie_hex))) { + XCECD_WORKER_LOG(XCE_LOG_ERROR, + "Failed to generate the X11 auth cookie for job %s", + jobidStr); + return -1; + } + + /* The listen socket stays bound for the job's lifetime; the accept + * handshake for the app's X connection must already succeed before the + * apiserver relay dials (X11 needs no protocol on accept). */ + int fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { + XCECD_WORKER_LOG(XCE_LOG_ERROR, "X11 forwarding socket failed: %s", + strerror(errno)); + return -1; + } + int one = 1; + setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)); + /* accept() must never block: the main loop shares this fd with the uv + * loop and a 1 ms cadence, so a blocking accept would freeze the whole + * worker (timers included) the moment the backlog runs dry */ + int fl = fcntl(fd, F_GETFL, 0); + if (fl < 0 || fcntl(fd, F_SETFL, fl | O_NONBLOCK) != 0) { + XCECD_WORKER_LOG(XCE_LOG_ERROR, + "X11 forwarding listener fcntl failed: %s", + strerror(errno)); + close(fd); + return -1; + } + + struct sockaddr_in addr; + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(x11_fwd.x11_port); + addr.sin_addr.s_addr = INADDR_ANY; + if (bind(fd, (struct sockaddr*)&addr, sizeof(addr)) != 0 || listen(fd, 8) != 0) { + XCECD_WORKER_LOG(XCE_LOG_ERROR, + "X11 forwarding listen on port %d failed: %s", + x11_fwd.x11_port, strerror(errno)); + close(fd); + return -1; + } + x11_fwd.listen_fd = fd; + + if (x11_install_cookie(jobid) != 0) { + x11_stop_forwarding(); + return -1; + } + + if (export_x11_display_env(x11_fwd.x11_port) != 0) { + x11_stop_forwarding(); + return -1; + } + + XCECD_WORKER_LOG(XCE_LOG_INFO, + "X11 forwarding ready for job %s: port %d, display :%d, " + "authority %s", + jobidStr, x11_fwd.x11_port, x11_fwd.x11_port - 6000, + x11_fwd.auth_file); + return 0; +} + +static void x11_stop_forwarding() +{ + if (x11_fwd.listen_fd >= 0) { + close(x11_fwd.listen_fd); + x11_fwd.listen_fd = -1; + } + for (int i = 0; i < 2; i++) { + if (x11_fwd.legs[i].fd >= 0) { + close(x11_fwd.legs[i].fd); + x11_fwd.legs[i].fd = -1; + } + x11_fwd.legs[i].len = 0; + } + x11_remove_cookie(); +} + +/* One pump step for a bridged connection: drain its stash toward the peer + * and pull new bytes into it. Non-blocking; short counts and EAGAIN are + * normal -- the stash absorbs bursts the peer cannot take right away, and + * a full stash applies backpressure (we simply stop recv'ing until the + * peer drains it) instead of dropping or killing the session. + * Returns false once the peer is gone (EOF or error), which ends the + * session: X11 has no notion of a half-closed connection, so when either + * leg dies the whole bridge dies with it (the job then exits on IO error, + * or the relay teardown reports the end to the CLI). */ +static bool x11_pump_leg(x11_leg_t* leg, x11_leg_t* peer) +{ + if (leg->fd < 0) { + return true; + } + + /* flush stashed bytes toward the peer first */ + if (leg->len > 0) { + ssize_t n = send(peer->fd, leg->stash, leg->len, + MSG_DONTWAIT | MSG_NOSIGNAL); + if (n < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) { + XCECD_WORKER_LOG(XCE_LOG_DEBUG, + "X11 bridge leg %d send failed: %s", leg->fd, + strerror(errno)); + return false; + } + } else if (n > 0) { + if (n < leg->len) { + memmove(leg->stash, leg->stash + n, leg->len - n); + } + leg->len -= n; + } + } + + char buf[8192]; + for (;;) { + if (leg->len >= X11_STASH_SIZE) { + return true; /* backpressure: wait for the peer to drain */ + } + ssize_t n = recv(leg->fd, buf, sizeof(buf), MSG_DONTWAIT); + if (n < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) { + return true; + } + XCECD_WORKER_LOG(XCE_LOG_DEBUG, "X11 bridge leg %d recv failed: %s", + leg->fd, strerror(errno)); + return false; + } + if (n == 0) { + XCECD_WORKER_LOG(XCE_LOG_DEBUG, "X11 bridge leg %d closed by peer", + leg->fd); + return false; + } + int off = 0; + while (off < n) { + int room = X11_STASH_SIZE - leg->len; + if (room <= 0) { + break; /* stash full -- drain on a later tick */ + } + int take = (n - off) < room ? (n - off) : room; + memcpy(leg->stash + leg->len, buf + off, take); + leg->len += take; + off += take; + + ssize_t w = send(peer->fd, leg->stash, leg->len, + MSG_DONTWAIT | MSG_NOSIGNAL); + if (w < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) { + continue; + } + XCECD_WORKER_LOG(XCE_LOG_DEBUG, + "X11 bridge leg %d send failed: %s", leg->fd, + strerror(errno)); + return false; + } + if (w < leg->len) { + memmove(leg->stash, leg->stash + w, leg->len - w); + } + leg->len -= w; + } + if (off < n) { + return true; /* stash filled mid-buffer -- drain on a later tick */ + } + } +} + +/* Classify an accepted connection by source address: the app runs on this + * host and its DISPLAY is ":N" (loopback), the apiserver relay dials the + * worker's non-loopback IP from the apiserver host. */ +static bool x11_conn_is_relay(int fd) +{ + struct sockaddr_in sa; + socklen_t sl = sizeof(sa); + if (getpeername(fd, (struct sockaddr*)&sa, &sl) != 0) { + return false; + } + uint32_t ip = ntohl(sa.sin_addr.s_addr); + return ip != INADDR_LOOPBACK; +} + +/* Non-blocking accept/pump step for the X11 listener, called from the main + * loop. v1 carries exactly one X connection: one app leg (loopback) plus + * one apiserver relay leg, bridged once both are in. Anything beyond that + * is refused loudly -- xim/clipboard-style secondary connections are out of + * scope and silent window-event loss would be worse than a clear failure + * (plan 2.4). Classification is by source address so connection arrival + * order cannot mispair two racing app connections with each other. */ +static void x11_accept_and_pump() +{ + if (x11_fwd.listen_fd < 0) { + return; + } + + bool bridged = (x11_fwd.legs[0].fd >= 0 && x11_fwd.legs[1].fd >= 0); + + for (;;) { + int fd = accept(x11_fwd.listen_fd, NULL, NULL); + if (fd < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR + && errno != ECONNABORTED) { + XCECD_WORKER_LOG(XCE_LOG_ERROR, + "X11 forwarding accept failed: %s", + strerror(errno)); + } + break; + } + int fl = fcntl(fd, F_GETFL, 0); + if (fl >= 0) { + fcntl(fd, F_SETFL, fl | O_NONBLOCK); + } + /* TCP_NODELAY: X11 turns interactive (cursor moves, key events) -- + * tiny packets must not wait for Nagle on either leg */ + int one = 1; + setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)); + + if (bridged) { + XCECD_WORKER_LOG(XCE_LOG_NOTICE, + "X11 job app opened a second X connection; " + "v1 supports one -- closing it"); + close(fd); + break; /* one refusal per tick keeps the loop responsive */ + } + + bool relay = x11_conn_is_relay(fd); + /* slot 0 = app leg, slot 1 = relay leg; a duplicate of a role we + * already hold (e.g. a second app racing in) is refused */ + int slot = relay ? 1 : 0; + if (x11_fwd.legs[slot].fd >= 0) { + XCECD_WORKER_LOG(XCE_LOG_NOTICE, + "X11 job app opened a second X connection; " + "v1 supports one -- closing it"); + close(fd); + break; + } + x11_fwd.legs[slot].fd = fd; + x11_fwd.legs[slot].relay = relay; + x11_fwd.legs[slot].len = 0; + x11_fwd.accepts++; + XCECD_WORKER_LOG(XCE_LOG_DEBUG, "X11 forwarding %s connection accepted", + relay ? "relay" : "app"); + + if (x11_fwd.legs[0].fd >= 0 && x11_fwd.legs[1].fd >= 0) { + bridged = true; + x11_fwd.saw_client = true; + x11_no_client_timer_stop(); + XCECD_WORKER_LOG(XCE_LOG_INFO, + "X11 bridge established on port %d", + x11_fwd.x11_port); + } + } + + if (bridged) { + if (!x11_pump_leg(&x11_fwd.legs[0], &x11_fwd.legs[1])) { + x11_stop_forwarding(); + return; + } + if (!x11_pump_leg(&x11_fwd.legs[1], &x11_fwd.legs[0])) { + x11_stop_forwarding(); + } + } +} + +static void x11_no_client_callback(uv_timer_t* timer) +{ + (void)timer; + + if (worker_exit_flag + || XJOB != mydata.transfer_info->interactive_mode + || 1 != mydata.transfer_info->x11_mode) { + return; + } + if (x11_fwd.saw_client) { + /* an X connection arrived: the session belongs to the app now */ + x11_no_client_timer_stop(); + return; + } + + pid_t pid = uv_proc_get_pid(&uv_proc_manager); + if (pid <= 0) { + x11_no_client_timer_stop(); + return; /* process_exit_callback owns the job */ + } + + char jobidStr[64]; + fmt_jobid(jobidStr, sizeof(jobidStr), jobid); + XCECD_WORKER_LOG(XCE_LOG_NOTICE, + "X11 job %s: no X client connected within %d s, reaping " + "the job", + jobidStr, (int)(X11_NO_CLIENT_MS / 1000)); + + tty_reap_reason = "No X11 client connected within 60 s"; + x11_no_client_timer_stop(); + /* the listener is not the app's parent: killing the job tree directly + * lands us in the regular process_exit_callback path */ + signal_children_recursive(SIGKILL, pid); + kill(pid, SIGKILL); +} + +static void x11_no_client_timer_start() +{ + if (!x11_fwd.timer_ready) { + return; + } + uv_timer_start(&x11_fwd.no_client_timer, x11_no_client_callback, + X11_NO_CLIENT_MS, X11_NO_CLIENT_MS); +} + +static void x11_no_client_timer_stop() +{ + if (x11_fwd.timer_ready) { + uv_timer_stop(&x11_fwd.no_client_timer); + } +} + static int xjob_start_vnc() { char jobidStr[64]; @@ -1090,8 +1589,19 @@ static void worker_job_callback(job_assignment_t* info) "Parsed exdata for job %s: interactive=%d", jobidStr, mydata.transfer_info->interactive_mode); - int job_port - = check_port(port_info, mydata.transfer_info->interactive_mode); + /* X11-forwarding jobs are interactive=XJOB with exdata x11=1: they take + * their port from the forwarding segment (6100-6199) instead of the VNC + * one, and generate_cmd passes the raw command through (the job needs no + * ttyd wrapper). */ + int x11_backend = (mydata.transfer_info->interactive_mode == XJOB) + && (1 == mydata.transfer_info->x11_mode); + + int job_port; + if (x11_backend) { + job_port = find_available_port(X11_START_PORT, X11_END_PORT); + } else { + job_port = check_port(port_info, mydata.transfer_info->interactive_mode); + } if (job_port < 0) { XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to check port availability for job %s", @@ -1102,6 +1612,8 @@ static void worker_job_callback(job_assignment_t* info) || mydata.transfer_info->interactive_mode == DISPLAY) { port_info.tty_port = job_port; } else if (mydata.transfer_info->interactive_mode == XJOB) { + /* X11 forwarding and VNC share the xjob_port slot; the backend + * branch decides what listens on it */ port_info.xjob_port = job_port; } XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Assigned port %d for job %s", @@ -1114,8 +1626,13 @@ static void worker_job_callback(job_assignment_t* info) exit(1); } - job_cmd = generate_cmd((char*)info->cmd, port_info, - mydata.transfer_info->interactive_mode); + if (x11_backend) { + /* no ttyd wrapper: the command runs directly with DISPLAY inherited */ + job_cmd = strdup((char*)info->cmd); + } else { + job_cmd = generate_cmd((char*)info->cmd, port_info, + mydata.transfer_info->interactive_mode); + } /* bug-1936: run pre-exec and the job in the submission's cwd, with HOME * = its exec home; before apply_env_vars() so an explicit --env HOME @@ -1139,7 +1656,33 @@ static void worker_job_callback(job_assignment_t* info) * for the next job (previously exit(1) here took down the whole * per-job worker with no user-visible reason). */ if (XJOB == mydata.transfer_info->interactive_mode) { - if (xjob_start_vnc() != 0) { + if (1 == mydata.transfer_info->x11_mode) { + /* X11 forwarding backend (xcecli -I -X): listen for the app's X + * connections and point DISPLAY at the listener instead of + * running VNC. */ + if (x11_start_forwarding() != 0) { + XCECD_WORKER_LOG(XCE_LOG_ERROR, + "Failed to start X11 forwarding for job %s", + jobidStr); + x11_stop_forwarding(); /* best-effort cleanup of partial start */ + + check_and_reconnect(); + + job_runtime_info.job_pid = 0; + job_runtime_info.job_status = JOB_FAILED; + job_runtime_info.worker_status = WORKER_EXITED; + job_runtime_info.exit_code = 1; + + ipc_worker_send_status( + &job_runtime_info, + "Failed to start X11 forwarding (check worker log)"); + worker_exit_flag = 1; + return_code = 1; + FREEUP(job_cmd); + ipc_worker_run(UV_RUN_NOWAIT); + return; + } + } else if (xjob_start_vnc() != 0) { XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to start VNC server for job %s", jobidStr); @@ -1204,6 +1747,13 @@ static void worker_job_callback(job_assignment_t* info) tty_attach_watchdog_start(); } + /* X11-forwarding jobs: reap when no X connection ever arrives (the + * app blocks inside XOpenDisplay and never exits on its own) */ + if (XJOB == mydata.transfer_info->interactive_mode + && 1 == mydata.transfer_info->x11_mode) { + x11_no_client_timer_start(); + } + if (mydata.transfer_info->interactive_mode > 0) { url = generate_url(port_info, mydata.transfer_info->interactive_mode); @@ -1213,6 +1763,23 @@ static void worker_job_callback(job_assignment_t* info) url); } + if (x11_backend && url) { + /* rewrite the XJOB URL ("ip::port") to the forwarding form + * the apiserver's xattach relay and the xcecli -X client + * expect ("x11://ip:port") -- the double colon is the VNC + * websockify convention and not a dialable address */ + char* dcolon = strstr(url, "::"); + char* x11url = calloc(1, URL_MAX_LENGTH); + if (x11url && dcolon) { + snprintf(x11url, URL_MAX_LENGTH, "x11://%.*s:%s", + (int)(dcolon - url), url, dcolon + 2); + FREEUP(url); + url = x11url; + } else { + FREEUP(x11url); + } + } + start_exdata = calloc(1, sizeof(start_exdata_t)); if (!start_exdata) { XCECD_WORKER_LOG( @@ -1221,6 +1788,13 @@ static void worker_job_callback(job_assignment_t* info) exit(1); } start_exdata->url = url; + /* -X jobs report the per-job X11 auth cookie next to the URL so + * the user's client can register it with their Xauthority */ + if (XJOB == mydata.transfer_info->interactive_mode + && 1 == mydata.transfer_info->x11_mode + && x11_fwd.cookie_hex[0] != '\0') { + start_exdata->x11cookie = strdup(x11_fwd.cookie_hex); + } start_exdata_info = generate_exdata(start_exdata); if (!start_exdata_info) { @@ -1252,7 +1826,10 @@ static void worker_job_callback(job_assignment_t* info) FREEUP(job_cmd); FREEUP(url); - FREEUP(start_exdata); + if (start_exdata) { + FREEUP(start_exdata->x11cookie); + FREEUP(start_exdata); + } } else { XCECD_WORKER_LOG(XCE_LOG_ERROR, @@ -1881,10 +2458,18 @@ int main(int argc, char* argv[]) uv_timer_init(uv_loop, &tty_attach_timer); tty_attach_timer_ready = true; + /* X11-forwarding no-client reaper (x11_no_client_callback) */ + uv_timer_init(uv_loop, &x11_fwd.no_client_timer); + x11_fwd.timer_ready = true; + do { ipc_worker_run(UV_RUN_NOWAIT); + /* X11 forwarding: accept the app's X connection and the apiserver + * relay's connection, then bridge them (non-blocking step) */ + x11_accept_and_pump(); + usleep(1000); } while (worker_exit_flag == 0); diff --git a/src/worker/testsuite/test_exdata_parse.c b/src/worker/testsuite/test_exdata_parse.c index ec5587ea..c9914078 100644 --- a/src/worker/testsuite/test_exdata_parse.c +++ b/src/worker/testsuite/test_exdata_parse.c @@ -149,6 +149,62 @@ static void test_apiserver_nofile_limit_parsed(void) TEST_ASSERT_EQUAL_UINT64(60, limits.run_limit); } +/* x11=1 (exdata key "x11", xcecli -I -X): selects the X11-forwarding + * backend for an XJOB job -- DISPLAY goes to the worker's forwarding port, + * no VNC. Must tolerate the string form like attach/interactive do. */ +static void test_x11_key_parsed(void) +{ + TEST_ASSERT_TRUE(parse_exdata( + "{\"interactive\":2,\"x11\":1}", &env, &tinfo, 1)); + TEST_ASSERT_EQUAL_INT(XJOB, tinfo.interactive_mode); + TEST_ASSERT_EQUAL_INT(1, tinfo.x11_mode); +} + +static void test_x11_key_string_form_parsed(void) +{ + TEST_ASSERT_TRUE(parse_exdata( + "{\"interactive\":\"2\",\"x11\":\"1\"}", &env, &tinfo, 1)); + TEST_ASSERT_EQUAL_INT(XJOB, tinfo.interactive_mode); + TEST_ASSERT_EQUAL_INT(1, tinfo.x11_mode); +} + +/* x11 on a non-XJOB job is recorded as parsed (the lifecycle branch keys + * off both fields together, so it stays inert there) but the parse itself + * must not fail the job. */ +static void test_x11_key_absent_defaults_zero(void) +{ + TEST_ASSERT_TRUE(parse_exdata( + "{\"interactive\":2}", &env, &tinfo, 1)); + TEST_ASSERT_EQUAL_INT(XJOB, tinfo.interactive_mode); + TEST_ASSERT_EQUAL_INT(0, tinfo.x11_mode); +} + +/* generate_exdata packs the -X job's auth cookie next to the URL; without + * a cookie (ttyd/VNC jobs) the exdata stays exactly {"url": ...}. */ +static void test_generate_exdata_x11cookie(void) +{ + start_exdata_t start = {.url = "x11://10.0.1.5:6100", + .x11cookie = "aabbccdd00112233"}; + char* json = generate_exdata(&start); + + TEST_ASSERT_NOT_NULL(json); + TEST_ASSERT_TRUE(strstr(json, "\"url\":\"x11://10.0.1.5:6100\"") != NULL); + TEST_ASSERT_TRUE( + strstr(json, "\"x11cookie\":\"aabbccdd00112233\"") != NULL); + free(json); +} + +static void test_generate_exdata_without_cookie(void) +{ + start_exdata_t start = {.url = "http://10.0.1.5:9000", + .x11cookie = NULL}; + char* json = generate_exdata(&start); + + TEST_ASSERT_NOT_NULL(json); + TEST_ASSERT_EQUAL_STRING("{\"url\":\"http://10.0.1.5:9000\"}", json); + free(json); +} + int main(void) { UNITY_BEGIN(); @@ -158,5 +214,10 @@ int main(void) RUN_TEST(test_both_keys_keep_first_list); RUN_TEST(test_apiserver_limits_parsed); RUN_TEST(test_apiserver_nofile_limit_parsed); + RUN_TEST(test_x11_key_parsed); + RUN_TEST(test_x11_key_string_form_parsed); + RUN_TEST(test_x11_key_absent_defaults_zero); + RUN_TEST(test_generate_exdata_x11cookie); + RUN_TEST(test_generate_exdata_without_cookie); return UNITY_END(); }