Repository navigation
Expand file tree
/
Copy pathserver.py
More file actions
3475 lines (2890 loc) · 146 KB
/
Copy pathserver.py
File metadata and controls
3475 lines (2890 loc) · 146 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import asyncio
import base64
import collections
import functools
import hashlib
import hmac
import json
import logging
import os
import re
import secrets
import shutil
import socket
import sqlite3
import subprocess
import sys
import threading
import time
import urllib.request
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from importlib.metadata import PackageNotFoundError, version
from logging.handlers import RotatingFileHandler
from pathlib import Path
from urllib.parse import quote, urlsplit
import segno
import uvicorn
from cryptography.exceptions import InvalidTag
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from . import agent_commands, detection, smart_recall
from . import agent_state as lifecycle
from . import agents, auth, context, db, gitgraph, integrations, plugins, remotes, share, shells, stats, team
from . import scheduler as sched
from . import e2e, ptyhost, relay, update
from .pty import PtyManager
log = logging.getLogger("shelldeck")
try:
VERSION = version("shelldeck")
except PackageNotFoundError:
VERSION = "dev"
STATIC = Path(__file__).parent / "static"
# Requests must target one of these hosts (blocks DNS rebinding). None = any host.
ALLOWED_HOSTS: set[str] | None = {"127.0.0.1", "localhost", "[::1]"}
SETTINGS_DEFAULTS = {
"default_shell": shells.default_kind(),
"wsl_distro": "",
"font_size": "13",
"theme": "dark",
"layout_mode": "tiled",
"project_tint": "on",
"terminal_theme": "default",
"font_family": "",
"editor": "vscode",
"agent_resume": "ask",
"ask_agent": "auto", # auto | claude | codex | gemini | devin: who writes `?` / `sd ask` commands
"ask_model": "", # that agent's model; empty = its smallest
"recall_agent": "off", # off | auto | claude | codex | gemini | devin: whose small model widens `sd recall --smart` # never | ask | auto: start a stored agent session again when its terminal comes back
}
# keep in sync with TERMINAL_THEMES in static/app.js
TERMINAL_THEMES = ("default", "dracula", "one-dark", "nord", "gruvbox-dark", "solarized-dark", "solarized-light", "github-light")
FONT_FAMILY = re.compile(r"[\w ,.'\"-]{0,120}")
manager: PtyManager = PtyManager() # a ptyhost.PtyClient under the supervisor (lifespan)
EXIT = {"code": 0} # sd-ui's exit code: tells the supervisor to stop, or to restart the UI (ptyhost.RESTART_UI)
sockets: dict[str, set[WebSocket]] = {}
readers: dict[str, asyncio.Task] = {}
alarm_sockets: set[WebSocket] = set()
clipboard = {"data": ""}
socket_owner: dict[WebSocket, str | None] = {} # open socket -> login session hash
last_cwd: dict[str, str] = {}
# sid -> the shell is at its prompt (shell integration's 133;B seen, no Enter since). Sent to a browser on connect:
# replayed scrollback has no prompt marks, so a page that just loaded can't tell by itself (the `?` menu needs it)
at_prompt: dict[str, bool] = {}
# sub-agent terminal -> its output since the last question check, and when it last printed
ask_buf: dict[str, str] = {}
last_out_at: dict[str, float] = {}
QUIET_S = 2.5 # a question counts once the sub-agent's screen stops changing
BOX = re.compile("[\u2500-\u257f]+") # rules and frames a TUI draws around its prompts
REASK_S = 120 # the same question redrawn within this long is not forwarded again
asked_before: dict[str, tuple[str, float]] = {} # sub-agent -> (last question forwarded, when)
# agent "needs you" detection (mirrors updateAgentStates in static/app.js): every terminal's last output,
# the start and text of its current burst, and each agent terminal's state: working | approval | idle
out_at: dict[str, float] = {}
busy_since: dict[str, float] = {}
burst: dict[str, str] = {}
agent_state: dict[str, str] = {}
# server-owned lifecycle (agent_state.py): reports per terminal, the resolved status sent to clients, the agent
# kind and a generation that changes when a different agent process takes the terminal (pins waits)
reports = lifecycle.Registry()
agent_status: dict[str, dict] = {}
agent_kind: dict[str, tuple[str, int]] = {}
_generation = [0]
report_tokens: dict[str, str] = {} # terminal -> SHELLDECK_AGENT_REPORT_TOKEN of its current process
REPORT_HEADER = "X-Shelldeck-Report-Token"
# server events for `sd events subscribe` / GET /api/events (SSE): a bounded in-memory log; ids only grow
EVENT_TYPES = ("agent.detected", "agent.state", "agent.command", "agent.exited", "agent.session_updated", "handoff.created", "handoff.completed",
"integration.changed")
events_log: "collections.deque[dict]" = collections.deque(maxlen=500)
_event_seq = [0]
_event_waiters: list[asyncio.Event] = []
_state_changed: list[asyncio.Event] = [] # set and replaced on every status change; `sd agent wait` awaits it
AGENT_BUSY_S = 5 # output for at least this long counts as working (not a redraw or echo)
# a sub-agent's question relayed to its parent (_forward_question): on the parent's screen, not asking you
RELAYED = re.compile(r"\[shelldeck\]\s+Your\s+sub-agent[\s\S]*?really\s+their\s+call")
EXPO_PUSH = "https://exp.host/--/api/v2/push/send"
STOPPING = asyncio.Event() # set at shutdown: dying terminals then don't message their parents
background: set[asyncio.Task] = set() # fire-and-forget tasks, referenced so they aren't collected
# shell integration reports the working directory (see shelldeck/integration)
CWD_REPORT = re.compile(r"\x1b\]633;P;Cwd=([^\x07\x1b]+)(?:\x07|\x1b\\)")
def err(code: str, status: int = 400) -> JSONResponse:
return JSONResponse({"error": code}, status_code=status)
# The share: a cloudflared quick tunnel this server runs (share.py). _share holds the tunnel's public
# hostname, its one unused link token (None once used; kept so the host UI can show its QR) and when that
# expires, and one grant per browser that opened a link: sha256(cookie) -> {id, ip, agent, at,
# state: pending|ok|denied, session}. ponytail: the share runs until stopped, with no one watching;
# the UI shows a "Sharing" pill so it's never invisible.
_share: dict = {}
_share_lock = asyncio.Lock() # held while cloudflared starts: "starting", and a second start waits
_share_sent: dict = {"key": None} # the last share_state broadcast, so the watcher notices expiry
SHARE_COOKIE = "sd_share"
LINK_TTL = 600 # a link must be opened within 10 minutes, once
APPROVAL_TTL = 300 # a pending device the host didn't answer is dropped after 5 minutes
SELF_URL = {"url": "http://127.0.0.1:5455"} # where the tunnel points; set by run()
CLIENT_HEADER = "x-shelldeck-client" # "ios" asks for a remote login (see _login_response)
APP_AGENT = "ShelldeckApp/" # the iOS app's user agent
DEVICE_COOKIE = "sd_device" # a trusted device's token: passes the share gate with no link or approval
def _sha(value: str) -> str:
return hashlib.sha256(value.encode()).hexdigest()
def _share_on() -> bool:
return bool(_share)
def _share_link_ok(token: str) -> bool:
key = _share.get("token")
return bool(key) and time.time() < _share["link_until"] and hmac.compare_digest(token, key)
def _grant(conn) -> dict | None:
"""This browser's grant on the running share: from an opened link, or an allowed grant for a trusted device."""
if not _share:
return None
if (cookie := conn.cookies.get(SHARE_COOKIE)) and (g := _share["grants"].get(_sha(cookie))):
return g
if not (token := conn.cookies.get(DEVICE_COOKIE)) or not (d := db.trusted_device(_sha(token))):
return None
key = f"device:{d['id']}"
if key not in _share["grants"]: # once per share
db.touch_trusted_device(d["id"])
_share["grants"][key] = {"id": d["id"], "ip": _client(conn), "agent": d["name"], "at": time.time(), "state": "ok", "session": None}
return _share["grants"][key]
def _via_share(conn) -> bool:
return _share_on() and conn.headers.get("host", "").rsplit(":", 1)[0] == _share["host"]
def _trusted(conn) -> bool:
"""Same-origin check. Browsers always send Origin cross-site; the CLI sends none.
On the share tunnel's host only /share/ pages are open; the rest needs a host-approved grant."""
headers = conn.headers
host = headers.get("host", "")
if _via_share(conn):
g = _grant(conn)
if not (conn.url.path.startswith("/share/") or (g and g["state"] == "ok")):
return False
elif ALLOWED_HOSTS is not None and host.rsplit(":", 1)[0] not in ALLOWED_HOSTS:
return False
origin = headers.get("origin")
return not origin or urlsplit(origin).netloc == host
def _to_utc(iso: str | None) -> str | None:
if not iso:
return None
try:
return datetime.fromisoformat(iso).astimezone(timezone.utc).isoformat()
except ValueError:
return iso
def get_settings() -> dict:
s = {k: db.get_setting(k, v) for k, v in SETTINGS_DEFAULTS.items()}
if s["default_shell"] not in shells.kinds(): # e.g. a Windows db opened on Linux
s["default_shell"] = shells.default_kind()
return s
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
STOPPING.clear()
db.init_db()
if code := auth.init_auth():
log.warning("no password yet; remote setup code: %s", code)
await asyncio.to_thread(plugins.load, _plugins_off())
await sched.start()
sched.set_alarm_callback(_broadcast_alarm)
await sched.reminder_service.sync_all()
global manager
if addr := os.environ.get(ptyhost.ADDR_ENV): # under the supervisor: the shells live in sd-pty
manager = ptyhost.PtyClient(addr, os.environ.pop(ptyhost.TOKEN_ENV, ""))
manager.store = db.config_dir() / "scrollback"
await manager.connect(_host_terminal, _host_lost)
else:
manager.store = db.config_dir() / "scrollback"
manager.prune({s["id"] for s in db.list_sessions()})
saver = asyncio.create_task(_save_scrollback())
asker = asyncio.create_task(_watch_questions())
watcher = asyncio.create_task(_watch_share())
agent_watch = asyncio.create_task(_watch_agents())
updates = asyncio.create_task(_watch_updates())
# every start refreshes the shelldeck skill of each agent CLI on PATH (dev/test homes skip it);
# `sd install-skill --remove` turns this off
if "SHELLDECK_HOME" not in os.environ and db.get_setting("agent_skills", "on") == "on":
try:
await asyncio.to_thread(team.install)
except OSError:
log.warning("installing agent skills failed", exc_info=True)
log.info("shelldeck %s started", VERSION)
yield
saver.cancel()
asker.cancel()
watcher.cancel()
agent_watch.cancel()
updates.cancel()
STOPPING.set()
if _share:
await asyncio.to_thread(share.stop)
_share.clear()
stats.stop()
sched.shutdown()
for task in list(readers.values()):
task.cancel()
if not isinstance(manager, ptyhost.PtyClient): # sd-pty keeps its shells; the supervisor stops it
manager.save()
for sid in list(manager.procs):
manager.terminate(sid)
def _host_terminal(sid: str) -> None:
"""A terminal already running in sd-pty when this UI started (an update restarted it): pump it again."""
if token := manager.env.get(sid, {}).get("SHELLDECK_AGENT_REPORT_TOKEN"):
report_tokens[sid] = token
if sid not in readers:
readers[sid] = asyncio.get_running_loop().create_task(_pump(sid))
def _host_lost() -> None:
if not STOPPING.is_set():
log.error("lost the pty host; asking the supervisor for a full restart")
logging.shutdown()
os._exit(ptyhost.RESTART_ALL)
async def _save_scrollback() -> None:
# ponytail: periodic flush; a hard power-off loses at most the last 15s of output
while True:
await asyncio.sleep(15)
try:
await asyncio.to_thread(manager.save)
except OSError:
log.warning("saving scrollback failed", exc_info=True)
async def _watch_share() -> None:
"""Tell host browsers when the link or a pending device expires (every other change says so itself)."""
while True:
await asyncio.sleep(5)
if _share and _share_key(_share_state()) != _share_sent["key"]:
await _share_changed()
class RevalidatedStaticFiles(StaticFiles):
"""Browsers must revalidate (a cheap 304), or they keep running old JS after an upgrade."""
def file_response(self, *args, **kwargs):
response = super().file_response(*args, **kwargs)
response.headers["Cache-Control"] = "no-cache"
return response
app = FastAPI(lifespan=lifespan, title="shelldeck", version=VERSION)
app.mount("/static", RevalidatedStaticFiles(directory=STATIC), name="static")
AUTH_EXEMPT = ("/api/auth/", "/api/health")
LOOPBACK = {"127.0.0.1", "::1", "localhost", "testclient"}
def _client(conn) -> str:
return conn.client.host if conn.client else "?"
def _is_local(conn) -> bool:
"""A request from the host itself (not through a reverse proxy)."""
proxied = "x-forwarded-for" in conn.headers or "cf-connecting-ip" in conn.headers # e.g. `sd share`
return _client(conn) in LOOPBACK and not proxied
def _who(conn) -> tuple[str, str | None] | None:
"""("cli", None) for the host CLI, ("browser", session hash) for a login, else None."""
if auth.cli_ok(conn.headers.get(auth.TOKEN_HEADER)):
return ("cli", None)
h = auth.session_hash(conn.cookies.get(auth.COOKIE))
return ("browser", h) if h else None
# One device uses shelldeck at a time; the others stay signed in but idle until they log in again.
_active: dict[str, str | None] = {"h": None}
def _claim(h: str) -> bool:
"""True when browser session `h` may act: it is the active one, or no live session is.
Remote (iOS app) logins always may, and never take the slot."""
if auth.remote(h):
return True
if _active["h"] != h and auth.alive(_active["h"]):
return False
_active["h"] = h
return True
def _denied() -> JSONResponse:
return err("locked" if auth.has_password() else "setup_required", 401)
SECURITY_HEADERS = {
"X-Frame-Options": "DENY",
"Content-Security-Policy": "frame-ancestors 'none'",
"X-Content-Type-Options": "nosniff",
"Referrer-Policy": "no-referrer",
}
@app.middleware("http")
async def guard(request: Request, call_next):
response = await _guarded(request, call_next)
response.headers.update(SECURITY_HEADERS)
if _via_share(request): # browsers then never use plain http to the tunnel host (cookies, password)
response.headers["Strict-Transport-Security"] = "max-age=31536000"
if request.url.path.startswith("/api/"):
response.headers["Cache-Control"] = "no-store" # terminal output, history, devices
return response
async def _guarded(request: Request, call_next):
if not _trusted(request):
return err("forbidden_origin", 403)
path = request.url.path
request.state.session = None
# an integration inside a terminal (hook script, OpenCode plugin) holds only its terminal's report token;
# agent_report() checks it. Host-local only: never through a share tunnel or the network.
report_only = path == "/api/agent-reports" and request.method == "POST" and REPORT_HEADER in request.headers and _is_local(request)
if path.startswith("/api/") and not path.startswith(AUTH_EXEMPT) and not report_only:
who = _who(request)
if not who:
return _denied()
request.state.session = who[1]
if who[1] and not _claim(who[1]):
return err("in_use", 423)
if who[1] and request.method != "GET": # polling GETs are not activity
auth.touch(who[1])
return await call_next(request)
@app.exception_handler(Exception)
async def unhandled(request: Request, exc: Exception):
log.exception("unhandled error on %s %s", request.method, request.url.path)
return err("internal_error", 500)
@app.get("/")
async def index():
return FileResponse(STATIC / "index.html", headers={"Cache-Control": "no-cache"})
@app.get("/api/health")
async def health():
return {"app": "shelldeck", "version": VERSION}
_server: uvicorn.Server | None = None
@app.post("/api/shutdown")
async def shutdown():
if _server:
_server.should_exit = True
return {"status": "stopping"}
@app.get("/api/update")
async def update_status(force: bool = False):
"""Is a newer shelldeck out? how: 'ota' (this standalone install updates in place) or 'uv' (uv tool upgrade)."""
try:
info = await asyncio.to_thread(update.check, VERSION, force)
except (OSError, ValueError, KeyError) as e:
log.warning("update check failed: %s", e)
return err("update_check_failed", 503)
# in place needs the supervisor: only then do the shells outlive a restart of this process
return info | {"in_place": info["how"] == "ota" and isinstance(manager, ptyhost.PtyClient)}
@app.post("/api/update")
async def update_apply(request: Request):
"""Install the latest standalone release next to this one and restart sd-ui from it; the shells keep running."""
if not _host(request):
return err("host_only", 403)
if not isinstance(manager, ptyhost.PtyClient):
return err("not_supervised", 409)
try:
new = await asyncio.to_thread(update.apply, VERSION)
except (OSError, ValueError, KeyError, RuntimeError) as e:
log.warning("update failed: %s", e)
return JSONResponse({"error": "update_failed", "detail": str(e)}, status_code=500)
log.info("updated to %s; restarting the UI", new)
_restart_ui()
return {"status": "restarting", "version": new}
@app.post("/api/restart")
async def restart_ui(request: Request):
"""Host only: restart sd-ui (new code, same shells)."""
if not _host(request):
return err("host_only", 403)
if not isinstance(manager, ptyhost.PtyClient):
return err("not_supervised", 409)
_restart_ui()
return {"status": "restarting"}
def _restart_ui() -> None:
EXIT["code"] = ptyhost.RESTART_UI
asyncio.get_running_loop().call_later(0.3, lambda: setattr(_server, "should_exit", True))
async def _watch_updates() -> None:
"""Tell host browsers once per new version (the page offers Update, or the upgrade command)."""
told = ""
while True:
await asyncio.sleep(60)
try:
info = await asyncio.to_thread(update.check, VERSION)
except (OSError, ValueError, KeyError):
info = {}
if info.get("available") and info["latest"] != told:
told = info["latest"]
await _broadcast({"type": "update", **info, "in_place": info["how"] == "ota" and isinstance(manager, ptyhost.PtyClient)}, host_only=True)
await asyncio.sleep(update.CHECK_S)
# ---------------------------------------------------------------------------
# Auth
# ---------------------------------------------------------------------------
def _via(request: Request) -> str:
if _via_share(request):
return f"share:{_share['host']}"
return "local" if _is_local(request) else "network"
async def _login_response(request: Request, body: dict, res=None):
"""Start a login session (cookie on `res`, JSON body by default). It becomes the active device:
this browser's previous login is replaced and every other device goes idle. A remote login (the
iOS app through an approved share grant or from the network) takes nothing over: the host browser stays in use."""
if old := auth.session_hash(request.cookies.get(auth.COOKIE)):
auth.end_session(old)
g = _grant(request) if _via_share(request) else None
app = request.headers.get(CLIENT_HEADER, "").lower() == "ios" or APP_AGENT in request.headers.get("user-agent", "")
# ponytail: "app" is only a claim, so a LAN browser could fake it; it still needs the password, and remote only skips takeover
remote = app and (g["state"] == "ok" if g else not _via_share(request) and not _is_local(request))
token = auth.new_session(f"{_client(request)} {request.headers.get('user-agent', '')}", _via(request),
"remote" if remote else "browser")
h = auth.session_hash(token)
if g:
g["session"] = h # revoking this login also voids the grant
if not remote:
_active["h"] = h
await _close_session_sockets(lambda owner: owner != h and not auth.remote(owner), IN_USE)
res = res or JSONResponse(body)
res.set_cookie(
auth.COOKIE, token, httponly=True, samesite="strict", path="/",
# through the tunnel sd-ui sees http (cloudflared -> loopback), but the browser is on https
secure=request.url.scheme == "https" or _via_share(request), max_age=30 * 24 * 3600,
)
return res
def _too_many(request: Request) -> JSONResponse | None:
wait = auth.blocked_for(_client(request))
if wait:
return JSONResponse({"error": "too_many_attempts", "retry_after": wait}, status_code=429, headers={"Retry-After": str(wait)})
return None
@app.get("/api/auth/status")
async def auth_status(request: Request):
who = _who(request)
if who and who[1]:
auth.touch(who[1])
return {
"has_password": auth.has_password(),
"setup_code_required": not auth.has_password() and not _is_local(request),
"authenticated": bool(who),
"in_use_elsewhere": bool(who and who[1] and not auth.remote(who[1]) and who[1] != _active["h"] and auth.alive(_active["h"])),
"idle_timeout": auth.LOCK_TIMEOUT_SECONDS,
"min_length": auth.MIN_PASSWORD,
}
@app.post("/api/auth/setup")
async def auth_setup(request: Request, payload: dict):
"""First run: create the password. Remote clients also need the code printed on the host."""
if auth.has_password():
return err("password_already_set", 409)
if blocked := _too_many(request):
return blocked
if not _is_local(request):
ok = hmac.compare_digest(str(payload.get("code", "")).strip().lower(), (auth.setup_code() or "").lower())
auth.record_attempt(_client(request), ok)
if not ok:
return err("invalid_setup_code", 403)
password, confirm = str(payload.get("password", "")), str(payload.get("confirm", ""))
if problem := auth.validate_new(password, confirm):
return err(problem)
auth.set_password(password)
log.info("password created from %s", _client(request))
return await _login_response(request, {"status": "ok"})
@app.post("/api/auth/login")
async def auth_login(request: Request, payload: dict):
if not auth.has_password():
return err("setup_required", 401)
if blocked := _too_many(request):
return blocked
ok = await asyncio.to_thread(auth.check_password, str(payload.get("password", "")))
auth.record_attempt(_client(request), ok)
if not ok:
log.warning("failed login from %s", _client(request))
return err("invalid_password", 401)
return await _login_response(request, {"status": "ok"})
_login_links: dict[str, float] = {} # one-time code -> expiry
@app.post("/api/auth/login-link")
async def auth_login_link(request: Request):
"""Host CLI only: a one-time URL that logs a browser in (emergency unlock)."""
if not auth.cli_ok(request.headers.get(auth.TOKEN_HEADER)):
return err("host_cli_only", 403)
code = secrets.token_urlsafe(24)
_login_links[code] = time.time() + 300
return {"path": f"/api/auth/link/{code}"}
def _host(request: Request) -> bool:
"""The host: its CLI, or a browser on this machine that isn't coming through the tunnel."""
return auth.cli_ok(request.headers.get(auth.TOKEN_HEADER)) or (_is_local(request) and not _via_share(request))
@functools.lru_cache(maxsize=4)
def _qr(url: str) -> str:
# omitsize: a viewBox instead of a fixed size, so CSS scales it (a fixed-size svg in a smaller box got cut off)
return segno.make(url, error="l").svg_inline(scale=5, border=2, light="#fff", omitsize=True)
def _share_state() -> dict:
"""ShareState (see plans/ios-app.md, API contract)."""
host, token = _share.get("host"), _share.get("token")
link = None
if token and time.time() < _share["link_until"]:
# the fragment never leaves the scanning device: the app pins the end-to-end key from it (e2e.py)
url = f"https://{host}/share/{token}#k={e2e.public()}"
link = {"path": f"/share/{token}", "url": url, "expires_at": int(_share["link_until"]), "qr_svg": _qr(url)}
return {
"sharing": bool(_share), "starting": _share_lock.locked(), "url": f"https://{host}" if host else None,
"link": link, "pending": _pending(), "strong_password": auth.strong_password(), "cloudflared": bool(share.command()),
"terms_accepted": auth.terms_accepted(share.TERMS_VERSION), "terms": share.TERMS,
}
def _share_key(state: dict) -> tuple:
return state["sharing"], state["starting"], bool(state["link"]), tuple(p["id"] for p in state["pending"])
async def _share_changed() -> None:
"""Every host browser gets the new ShareState."""
state = _share_state()
_share_sent["key"] = _share_key(state)
await _broadcast({"type": "share_state", "state": state}, host_only=True)
def _pending() -> list[dict]:
"""Devices waiting for the host; stale ones are dropped."""
grants, now = _share.get("grants", {}), time.time()
for k in [k for k, g in grants.items() if g["state"] == "pending" and now - g["at"] > APPROVAL_TTL]:
del grants[k]
return [{"id": g["id"], "ip": g["ip"], "agent": g["agent"], "at": g["at"], "code": g.get("code")}
for g in grants.values() if g["state"] == "pending"]
@app.post("/api/share")
async def share_start(request: Request):
"""Host only: start the tunnel (or return the running share) with a fresh link."""
if not _host(request):
return err("host_only", 403)
if not auth.strong_password():
return err("weak_password", 409)
if not auth.terms_accepted(share.TERMS_VERSION):
return err("terms_required", 409)
if not share.command():
return JSONResponse({"error": "no_cloudflared", "hint": share.HINT}, status_code=404)
problem = None
async with _share_lock:
if not _share:
await _share_changed() # starting
loop = asyncio.get_running_loop()
def exited(host: str) -> None: # cloudflared's reader thread
loop.call_soon_threadsafe(_spawn, _tunnel_exited(host))
try:
host = await asyncio.to_thread(share.start, SELF_URL["url"], exited)
_share.update(host=host, grants={})
_new_share_link()
log.info("sharing via %s", host)
_spawn(_announce(host))
except TimeoutError:
problem = err("tunnel_timeout", 504)
except (OSError, RuntimeError):
log.warning("cloudflared failed to start", exc_info=True)
problem = err("tunnel_failed", 502)
await _share_changed()
return problem or _share_state()
def _spawn(coro) -> None:
t = asyncio.get_running_loop().create_task(coro)
background.add(t)
t.add_done_callback(background.discard)
async def _announce(host: str) -> None:
"""Trusted devices learn the new share URL now, and again every REPEAT_S while this share runs."""
while _share.get("host") == host:
if devices := db.list_trusted_devices():
await asyncio.to_thread(relay.publish, devices, f"https://{host}")
await asyncio.sleep(relay.REPEAT_S)
@app.post("/api/trusted")
async def trust_device(request: Request, payload: dict | None = None):
"""The iOS app, logged in (over the LAN or an approved share), makes itself a trusted device. The token,
ntfy topic and key are returned once; the server keeps the token's hash."""
if not auth.remote(request.state.session):
return err("app_only", 403)
token = secrets.token_urlsafe(32)
topic, key = relay.new_device()
d = db.add_trusted_device(str((payload or {}).get("name") or "iPhone"), _sha(token), topic, key)
log.info("trusted device %s (%s) added from %s", d["id"], d["name"], _client(request))
await _broadcast({"type": "trusted", "device": {"id": d["id"], "name": d["name"]}}, host_only=True)
if _share.get("host"):
await asyncio.to_thread(relay.publish, [d], f"https://{_share['host']}")
return {"id": d["id"], "token": token, "topic": topic, "key": key, "relay": relay.NTFY, "cookie": DEVICE_COOKIE,
"pc": socket.gethostname()}
@app.get("/api/trusted/me")
async def trusted_me(request: Request):
"""Is this request's device token still trusted? The app asks before enrolling again, so it reuses its trust."""
d = db.trusted_device(_sha(token)) if (token := request.cookies.get(DEVICE_COOKIE)) else None
return {"trusted": bool(d), "id": d["id"] if d else None}
@app.get("/api/trusted")
async def trusted_list():
return {"devices": [{k: d[k] for k in ("id", "name", "created_at", "last_seen")} for d in db.list_trusted_devices()]}
@app.delete("/api/trusted/{device_id}")
async def trusted_delete(request: Request, device_id: str):
"""Untrust a device (not from the tunnel). Its logins through the share end too."""
if _via_share(request):
return err("host_only", 403)
if not db.delete_trusted_device(device_id):
return err("not_found", 404)
if g := _share.get("grants", {}).pop(f"device:{device_id}", None):
await _revoke([g["session"]] if g["session"] else [])
return {"status": "ok"}
async def _tunnel_exited(host: str) -> None:
if _share.get("host") == host:
log.warning("cloudflared exited; sharing stopped")
await _share_stop()
def _new_share_link() -> None:
"""One link at a time: a new one voids the unused old one."""
_share.update(token=secrets.token_urlsafe(32), link_until=time.time() + LINK_TTL)
@app.get("/api/share")
async def share_get(request: Request):
"""Host only: ShareState."""
if not _host(request):
return err("host_only", 403)
return _share_state()
@app.post("/api/share/terms")
async def share_terms(request: Request, payload: dict):
"""Host only: accept the share terms (asked once, again when TERMS_VERSION changes)."""
if not _host(request):
return err("host_only", 403)
if payload.get("accept") is not True:
return err("accept_required")
auth.accept_terms(share.TERMS_VERSION)
return _share_state()
@app.post("/api/share/link")
async def share_link(request: Request):
"""Host only: a fresh link for another device on the running share."""
if not _host(request):
return err("host_only", 403)
if not _share:
return err("not_sharing", 404)
_new_share_link()
await _share_changed()
return _share_state()
@app.post("/api/share/decide")
async def share_decide(request: Request, payload: dict):
"""Allow or deny a device that opened a link: the host CLI, or any logged-in browser not on the tunnel."""
if _via_share(request):
return err("host_only", 403)
_pending()
g = next((g for g in _share.get("grants", {}).values() if g["id"] == payload.get("id")), None)
if not g or g["state"] != "pending":
return err("not_found", 404)
g["state"] = "ok" if payload.get("allow") is True else "denied"
log.info("share device %s from %s %s", g["id"], g["ip"], "allowed" if g["state"] == "ok" else "denied")
await _share_changed()
return {"state": g["state"]}
@app.delete("/api/share")
async def share_stop(request: Request):
if not _host(request):
return err("host_only", 403)
await _share_stop()
return {"status": "ok"}
async def _share_stop() -> None:
"""Stop the tunnel and sign out every browser that logged in through it."""
if not _share:
return
gone = db.delete_auth_sessions_via(f"share:{_share['host']}")
_share.clear()
await asyncio.to_thread(share.stop)
await _revoke(gone)
await _share_changed()
_WAIT_PAGE = """<!doctype html><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>shelldeck</title><meta name="referrer" content="no-referrer">
<style>body{margin:0;min-height:100vh;display:grid;place-items:center;background:#09090b;color:#e4e4e7;
font:16px/1.5 Inter,"Segoe UI",system-ui,sans-serif;text-align:center;padding:24px;box-sizing:border-box}
p{max-width:28em;margin:.4em auto}b{color:#a78bfa}.m{color:#a1a1aa;font-size:14px}
.code{font:600 40px/1.2 ui-monospace,Consolas,monospace;letter-spacing:.3em;color:#e4e4e7}</style>
<main><p><b>shelldeck</b></p><p id="s">Waiting for the host to allow this device…</p>
<p id="c" class="code"></p>
<p class="m" id="m">Allow it in shelldeck on the host, or in the <code>sd share</code> window, if it shows this code.</p></main>
<script>
const say = (s, m) => { document.getElementById("s").textContent = s; document.getElementById("m").textContent = m;
document.getElementById("c").textContent = ""; };
async function poll() {
let st = "none";
try {
const r = await (await fetch("/share/status", { cache: "no-store" })).json();
st = r.state;
if (r.code) document.getElementById("c").textContent = r.code;
} catch {}
if (st === "ok") return location.replace("/");
if (st === "denied") return say("The host denied this device.", "Ask for a new link if that was a mistake.");
if (st !== "pending") return say("This request expired.", "Ask the host for a new link.");
setTimeout(poll, 2000);
}
poll();
</script>"""
def _share_page(body: str, status: int = 200):
return HTMLResponse(body, status_code=status, headers={"Cache-Control": "no-store"})
@app.get("/share/status")
async def share_status(request: Request):
"""The waiting page's poll: pending, ok, denied, or none."""
g = _grant(request) if _via_share(request) else None
body = {"state": g["state"] if g else "none"}
if g and g["state"] == "pending":
body["code"] = g.get("code") # only this browser's own code: the host compares it before allowing
return JSONResponse(body, headers={"Cache-Control": "no-store"})
@app.get("/share/{token}")
async def share_open(request: Request, token: str):
"""The QR link: works once, within LINK_TTL. It makes a pending grant that the host must allow (in a host
browser or the `sd share` window); after that the usual password login takes over."""
if not _via_share(request):
return err("link_expired", 403)
if g := _grant(request): # this browser already opened a link (e.g. a reload)
return RedirectResponse("/", status_code=303) if g["state"] == "ok" else _share_page(_WAIT_PAGE)
if not _share_link_ok(token):
return _share_page(_WAIT_PAGE.replace("poll();", 'say("This link has expired or was already used.", "Ask the host for a new one.");'), 403)
_share["token"] = None # one use
cookie = secrets.token_urlsafe(32)
_share["grants"][_sha(cookie)] = {
"id": secrets.token_hex(4), "ip": _client(request), "agent": request.headers.get("user-agent", "")[:200],
"at": time.time(), "state": "pending", "session": None,
# shown on the device and next to Allow: whoever saw the QR and opened it first gets a different code
"code": f"{secrets.randbelow(10000):04d}",
}
log.info("share link opened from %s; waiting for the host", _client(request))
await _share_changed()
res = _share_page(_WAIT_PAGE)
res.set_cookie(SHARE_COOKIE, cookie, httponly=True, samesite="lax", secure=True, path="/", max_age=24 * 3600)
return res
@app.get("/api/auth/link/{code}")
async def auth_use_link(request: Request, code: str):
expiry = _login_links.pop(code, 0)
if expiry < time.time() or not auth.has_password():
return err("link_expired", 403)
return await _login_response(request, {}, RedirectResponse("/", status_code=303))
@app.post("/api/auth/logout")
async def auth_logout(request: Request):
"""Lock this browser: end its session and close its sockets."""
h = auth.session_hash(request.cookies.get(auth.COOKIE))
if h:
auth.end_session(h)
await _close_session_sockets(lambda owner: owner == h)
res = JSONResponse({"status": "locked"})
res.delete_cookie(auth.COOKIE, path="/")
return res
@app.get("/api/devices")
async def devices(request: Request):
"""Browsers signed in: where from, when, and which one is in use."""
online: dict[str, int] = {}
for owner in socket_owner.values():
if owner:
online[owner] = online.get(owner, 0) + 1
rows = []
for r in db.list_auth_sessions():
h = r["token_hash"]
if not auth.alive(h):
continue
ip, _, agent = (r["client"] or "").partition(" ")
rows.append({
"id": h, "ip": ip, "agent": agent, "via": r["via"] or "local", "kind": r["kind"] or "browser",
"created_at": r["created_at"], "last_seen": r["last_seen"],
"active": h == _active["h"], "current": h == request.state.session, "sockets": online.get(h, 0),
})
on = _share_on()
return {"devices": rows, "share": _share["host"] if on else None, "pending": _pending() if on and not _via_share(request) else []}
@app.delete("/api/devices/{device_id}")
async def revoke_device(request: Request, device_id: str):
"""Sign one browser out, or with id "others" every browser but this one."""
if device_id == "others":
return await _revoke([d["token_hash"] for d in db.list_auth_sessions() if d["token_hash"] != request.state.session])
return await _revoke([device_id])
async def _revoke(hashes: list[str]) -> dict:
grants = _share.get("grants", {})
for k in [k for k, g in grants.items() if g["session"] in hashes]:
del grants[k] # a revoked share device needs a new link and approval
db.delete_push_tokens(hashes)
for h in hashes:
auth.end_session(h)
if _active["h"] == h:
_active["h"] = None
await _close_session_sockets(lambda owner: owner in hashes)
return {"revoked": len(hashes)}
@app.put("/api/auth/password")
async def auth_password(request: Request, payload: dict):
"""Change the password (logged-in browser only). Signs out every other browser."""
h = auth.session_hash(request.cookies.get(auth.COOKIE))
if not h:
return _denied()
if blocked := _too_many(request):
return blocked
ok = await asyncio.to_thread(auth.check_password, str(payload.get("current", "")))
auth.record_attempt(_client(request), ok)
if not ok:
return err("wrong_current_password", 403)
password, confirm = str(payload.get("password", "")), str(payload.get("confirm", ""))
if problem := auth.validate_new(password, confirm):
return err(problem)
auth.set_password(password)
auth.end_other_sessions(keep=h)
await _close_session_sockets(lambda owner: owner != h)
log.info("password changed from %s", _client(request))
return {"status": "ok"}
# ---------------------------------------------------------------------------
# Settings
# ---------------------------------------------------------------------------
@app.get("/api/settings")
async def read_settings():
return get_settings()
@app.put("/api/settings")
async def write_settings(payload: dict):
for key, value in payload.items():
if key not in SETTINGS_DEFAULTS:
return err(f"unknown_setting:{key}")
value = str(value).strip()
if key == "default_shell" and value not in shells.kinds():
return err("invalid_shell")
if key == "font_size" and not (value.isdigit() and 8 <= int(value) <= 32):
return err("invalid_font_size")
if key == "theme" and value not in ("dark", "light", "system"):
return err("invalid_theme")
if key == "layout_mode" and value not in ("tiled", "free"):
return err("invalid_layout_mode")
if key == "project_tint" and value not in ("on", "off"):
return err("invalid_project_tint")
if key == "terminal_theme" and value not in TERMINAL_THEMES:
return err("invalid_terminal_theme")
if key == "font_family" and not FONT_FAMILY.fullmatch(value):
return err("invalid_font_family")
if key == "editor" and value not in ("vscode", "system", "shelldeck"):
return err("invalid_editor")
if key == "agent_resume" and value not in ("never", "ask", "auto"):
return err("invalid_agent_resume")
if key == "recall_agent" and value not in ("off", "auto", *smart_recall.PRIORITY):
return err("invalid_recall_agent")
if key == "ask_agent" and value not in ("auto", *smart_recall.PRIORITY):
return err("invalid_ask_agent")
if key == "ask_model" and value and not team.SAFE_MODEL.match(value):
return err("invalid_ask_model")
for key, value in payload.items():
db.set_setting(key, str(value).strip())
return get_settings()
@app.get("/api/stats")
async def get_stats():
"""System CPU/RAM/GPU plus usage per running terminal (shell and its child processes)."""
return await asyncio.to_thread(stats.collect, _shell_pids())
@app.get("/api/sessions/{session_id}/processes")
async def session_processes(session_id: str):
"""The processes running in one terminal (below its shell), for sd ps."""
proc = manager.get(session_id)
if not proc or not proc.isalive() or not proc.pid:
return err("not_running", 409)