backend_server.py
7.94 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
"""Local backend process skeleton (U-007, extended in U-022): pure stdlib
`http.server`, no third-party deps (05 §5.7) -- this process is also where
the Graph Engine scheduler will live later (05 §5.4.1), but that's a
different Unit.
MVP scope: process start/stop, `c_pick_folder` (03-C §3.-1, now all three
platforms -- U-007 shipped macOS only via `osascript`; U-022 adds Windows
via a `powershell` FolderBrowserDialog script and Linux via `zenity`/
`kdialog`, whichever is installed), `c_project_create` (03-C §3.0, wired to
the U-002 `project_scaffold.create_project()` function that had no HTTP
route until now) for `v_project_setup`'s submit flow, and serving the U-008
frontend shell as static files (no build step, 05 §5.7: whitelisted by exact
path, not a generic directory walker, so there's no path-traversal surface
to reason about).
"""
import json
import os
import platform
import subprocess
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from src import paths, project_scaffold, schema
_FRONTEND_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend")
# U-008: exact-path whitelist, not a generic static-file server -- the
# frontend is 3 fixed files until PH3 adds real pages.
_STATIC_FILES = {
"/": ("index.html", "text/html; charset=utf-8"),
"/index.html": ("index.html", "text/html; charset=utf-8"),
"/app.js": ("app.js", "text/javascript; charset=utf-8"),
"/style.css": ("style.css", "text/css; charset=utf-8"),
}
class DialogUnavailableError(Exception):
"""DIALOG_UNAVAILABLE -- 03-C §3.-1: no native-dialog capability here."""
def pick_folder():
"""c_pick_folder (03-C §3.-1), all three platforms. Returns
{"selected_path": str} on success, {"selected_path": None} on user
cancel (not an error). Raises DialogUnavailableError if the platform's
native dialog tool itself fails or isn't installed."""
system = platform.system()
if system == "Darwin":
return _pick_folder_macos()
if system == "Windows":
return _pick_folder_windows()
return _pick_folder_linux()
def _pick_folder_macos():
script = 'POSIX path of (choose folder with prompt "Select a graphlemon project folder")'
result = subprocess.run(["osascript", "-e", script], capture_output=True, text=True)
if result.returncode != 0:
if "User canceled" in result.stderr or "(-128)" in result.stderr:
return {"selected_path": None}
raise DialogUnavailableError(result.stderr.strip())
return {"selected_path": result.stdout.strip()}
def _pick_folder_windows():
# .NET FolderBrowserDialog via a one-shot PowerShell script -- no
# cancel-vs-error signal other than empty stdout on a clean exit.
script = (
"Add-Type -AssemblyName System.Windows.Forms; "
"$f = New-Object System.Windows.Forms.FolderBrowserDialog; "
"if ($f.ShowDialog() -eq [System.Windows.Forms.DialogResult]::OK) { Write-Output $f.SelectedPath }"
)
try:
result = subprocess.run(
["powershell", "-NoProfile", "-NonInteractive", "-Command", script],
capture_output=True, text=True,
)
except FileNotFoundError as e:
raise DialogUnavailableError(str(e))
if result.returncode != 0:
raise DialogUnavailableError(result.stderr.strip())
return {"selected_path": result.stdout.strip() or None}
def _pick_folder_linux():
# zenity (GNOME) and kdialog (KDE) are the two dialogs actually present
# across mainstream Linux desktops; try whichever exists. Both exit
# non-zero on user cancel, indistinguishable here from "no folder
# chosen" -- both cases map to selected_path=None, not an error.
for cmd in (
["zenity", "--file-selection", "--directory"],
["kdialog", "--getexistingdirectory", os.path.expanduser("~")],
):
try:
result = subprocess.run(cmd, capture_output=True, text=True)
except FileNotFoundError:
continue
if result.returncode != 0:
return {"selected_path": None}
return {"selected_path": result.stdout.strip()}
raise DialogUnavailableError("neither zenity nor kdialog is installed")
class _Handler(BaseHTTPRequestHandler):
def do_GET(self):
entry = _STATIC_FILES.get(self.path)
if entry is None:
self._send_json(404, {"error": "NOT_FOUND"})
return
filename, content_type = entry
with open(os.path.join(_FRONTEND_DIR, filename), "rb") as f:
body = f.read()
self.send_response(200)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_POST(self):
if self.path == "/api/system/pick-folder":
self._handle_pick_folder()
elif self.path == "/api/projects":
self._handle_project_create()
else:
self._send_json(404, {"error": "NOT_FOUND"})
def _handle_pick_folder(self):
try:
self._send_json(200, pick_folder())
except DialogUnavailableError as e:
self._send_json(422, {"error": "DIALOG_UNAVAILABLE", "detail": str(e)})
def _handle_project_create(self):
length = int(self.headers.get("Content-Length", 0))
try:
payload = json.loads(self.rfile.read(length) or b"{}")
except json.JSONDecodeError:
self._send_json(400, {"error": "INVALID_INPUT", "detail": "body is not valid JSON"})
return
target_folder_path = payload.get("target_folder_path")
if not target_folder_path:
self._send_json(400, {"error": "INVALID_INPUT", "detail": "target_folder_path is required"})
return
try:
with self.server.conn_lock:
result = project_scaffold.create_project(
self.server.conn,
target_folder_path,
name=payload.get("name"),
init_git_if_missing=bool(payload.get("init_git_if_missing", False)),
)
except project_scaffold.PathNotFoundError as e:
self._send_json(404, {"error": "PATH_NOT_FOUND", "detail": str(e)})
except project_scaffold.PathNotWritableError as e:
self._send_json(403, {"error": "PATH_NOT_WRITABLE", "detail": str(e)})
except project_scaffold.GitInitDeclinedError as e:
self._send_json(409, {"error": "GIT_INIT_DECLINED", "detail": str(e)})
else:
self._send_json(200, result)
def _send_json(self, status, obj):
body = json.dumps(obj).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, format, *args):
pass # quiets test output; real request logging is a later Unit's concern
def _ensure_schema(conn):
exists = conn.execute(
"SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'project'"
).fetchone()
if exists is None:
schema.create_schema(conn)
class _Server(ThreadingHTTPServer):
def server_close(self):
super().server_close()
self.conn.close()
def create_server(host="127.0.0.1", port=0, db_path=None):
"""port=0 lets the OS assign a free port -- avoids collisions in tests
and lets multiple graphlemon instances run side by side if ever needed.
db_path defaults to paths.db_path() (U-003's platform_home, so it never
lives inside a target project's own repo); tests pass ":memory:" or a
tempfile so runs don't touch the real developer's ~/.graphlemon/."""
conn = schema.connect(db_path if db_path is not None else paths.db_path(), check_same_thread=False)
_ensure_schema(conn)
server = _Server((host, port), _Handler)
server.conn = conn
server.conn_lock = threading.Lock()
return server