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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions api/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,11 @@
HOST = os.getenv('HERMES_WEBUI_HOST', '127.0.0.1')
PORT = int(os.getenv('HERMES_WEBUI_PORT', '8787'))

# ── TLS/HTTPS config (optional, env-overridable) ────────────────────────────
TLS_CERT = os.getenv('HERMES_WEBUI_TLS_CERT', '').strip() or None
TLS_KEY = os.getenv('HERMES_WEBUI_TLS_KEY', '').strip() or None
TLS_ENABLED = TLS_CERT is not None and TLS_KEY is not None

# ── State directory (env-overridable, never inside repo) ──────────────────────
STATE_DIR = Path(os.getenv(
'HERMES_WEBUI_STATE_DIR',
Expand Down
20 changes: 17 additions & 3 deletions api/routes.py
Original file line number Diff line number Diff line change
Expand Up @@ -862,11 +862,25 @@ def _handle_list_dir(handler, parsed):
qs = parse_qs(parsed.query)
sid = qs.get('session_id', [''])[0]
if not sid: return bad(handler, 'session_id is required')
try: s = get_session(sid)
except KeyError: return bad(handler, 'Session not found', 404)
try:
s = get_session(sid)
workspace = s.workspace
except KeyError:
# Fallback for CLI sessions not loaded in WebUI memory
try:
cli_meta = None
for cs in get_cli_sessions():
if cs['session_id'] == sid:
cli_meta = cs
break
if not cli_meta:
return bad(handler, 'Session not found', 404)
workspace = cli_meta.get('workspace', '')
except Exception:
return bad(handler, 'Session not found', 404)
try:
return j(handler, {
'entries': list_dir(Path(s.workspace), qs.get('path', ['.'])[0]),
'entries': list_dir(Path(workspace), qs.get('path', ['.'])[0]),
'path': qs.get('path', ['.'])[0],
})
except (FileNotFoundError, ValueError) as e:
Expand Down
29 changes: 23 additions & 6 deletions api/updates.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,22 +64,32 @@ def _check_repo(path, name):
if not fetch_ok:
return {'name': name, 'behind': 0, 'error': 'fetch failed'}

branch = _detect_default_branch(path)
# Use the current branch's upstream tracking branch, not the repo default.
# This avoids false "N updates behind" alerts when the user is on a feature
# branch and master/main has moved forward with unrelated commits.
# If no upstream is set (brand-new local branch), fall back to the default branch.
upstream, ok = _run_git(['rev-parse', '--abbrev-ref', '@{upstream}'], path)
if ok and upstream:
# upstream is like "origin/feat/foo" — use it directly in rev-list
compare_ref = upstream
else:
branch = _detect_default_branch(path)
compare_ref = f'origin/{branch}'

# Count commits behind
out, ok = _run_git(['rev-list', '--count', f'HEAD..origin/{branch}'], path)
out, ok = _run_git(['rev-list', '--count', f'HEAD..{compare_ref}'], path)
behind = int(out) if ok and out.isdigit() else 0

# Get short SHAs for display
current, _ = _run_git(['rev-parse', '--short', 'HEAD'], path)
latest, _ = _run_git(['rev-parse', '--short', f'origin/{branch}'], path)
latest, _ = _run_git(['rev-parse', '--short', compare_ref], path)

return {
'name': name,
'behind': behind,
'current_sha': current,
'latest_sha': latest,
'branch': branch,
'branch': compare_ref,
}


Expand Down Expand Up @@ -129,7 +139,14 @@ def _apply_update_inner(target):
if path is None or not (path / '.git').exists():
return {'ok': False, 'message': 'Not a git repository'}

branch = _detect_default_branch(path)
# Use the current branch's upstream for pull, matching the behaviour
# of _check_repo. Falls back to default branch if no upstream is set.
upstream, ok = _run_git(['rev-parse', '--abbrev-ref', '@{upstream}'], path)
if ok and upstream:
compare_ref = upstream
else:
branch = _detect_default_branch(path)
compare_ref = f'origin/{branch}'

# Check for dirty working tree
status_out, _ = _run_git(['status', '--porcelain'], path)
Expand All @@ -141,7 +158,7 @@ def _apply_update_inner(target):
stashed = True

# Pull with ff-only (no merge commits)
pull_out, pull_ok = _run_git(['pull', '--ff-only', 'origin', branch], path, timeout=30)
pull_out, pull_ok = _run_git(['pull', '--ff-only', compare_ref], path, timeout=30)
if not pull_ok:
if stashed:
_run_git(['stash', 'pop'], path)
Expand Down
20 changes: 18 additions & 2 deletions server.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,10 +80,26 @@ def main() -> None:
SESSION_DIR.mkdir(parents=True, exist_ok=True)
DEFAULT_WORKSPACE.mkdir(parents=True, exist_ok=True)
httpd = ThreadingHTTPServer((HOST, PORT), Handler)
print(f' Hermes Web UI listening on http://{HOST}:{PORT}', flush=True)

# ── TLS/HTTPS setup (optional) ─────────────────────────────────────────
from api.config import TLS_ENABLED, TLS_CERT, TLS_KEY
scheme = 'https' if TLS_ENABLED else 'http'
if TLS_ENABLED:
try:
import ssl
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.minimum_version = ssl.TLSVersion.TLSv1_2
ctx.load_cert_chain(TLS_CERT, TLS_KEY)
httpd.socket = ctx.wrap_socket(httpd.socket, server_side=True)
print(f' TLS enabled: cert={TLS_CERT}, key={TLS_KEY}', flush=True)
except Exception as e:
print(f'[!!] WARNING: TLS setup failed ({e}), falling back to HTTP', flush=True)
scheme = 'http'

print(f' Hermes Web UI listening on {scheme}://{HOST}:{PORT}', flush=True)
if HOST == '127.0.0.1':
print(f' Remote access: ssh -N -L {PORT}:127.0.0.1:{PORT} <user>@<your-server>', flush=True)
print(f' Then open: http://localhost:{PORT}', flush=True)
print(f' Then open: {scheme}://localhost:{PORT}', flush=True)
print('', flush=True)
httpd.serve_forever()

Expand Down
214 changes: 214 additions & 0 deletions tests/test_tls_support.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,214 @@
"""
Tests for optional TLS/HTTPS support (HERMES_WEBUI_TLS_CERT / TLS_KEY).

Tests use a self-signed certificate generated at test time via openssl.
"""
import http.client
import json
import os
import ssl
import subprocess
import textwrap
import time
import tempfile
import unittest
from contextlib import suppress
from pathlib import Path

ROOT = Path(__file__).parent.parent


def _gen_test_cert(tmpdir: Path) -> tuple[str, str]:
"""Generate a self-signed cert and key pair for testing."""
cert = str(tmpdir / "test_cert.pem")
key = str(tmpdir / "test_key.pem")
subprocess.run(
["openssl", "req", "-x509", "-newkey", "rsa:2048",
"-keyout", key, "-out", cert, "-days", "1", "-nodes",
"-subj", "/CN=localhost"],
check=True, capture_output=True,
)
return cert, key


def _find_free_port() -> int:
import socket
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]


def _wait_for_server(host: str, port: int, use_ssl: bool = False,
timeout: float = 8.0) -> bool:
"""Poll until the server accepts a connection or times out."""
ctx = None
if use_ssl:
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
deadline = time.time() + timeout
while time.time() < deadline:
try:
if use_ssl:
c = http.client.HTTPSConnection(host, port, timeout=2, context=ctx)
else:
c = http.client.HTTPConnection(host, port, timeout=2)
c.request("GET", "/health")
resp = c.getresponse()
resp.read()
c.close()
return True
except Exception:
time.sleep(0.5)
return False


def _start_server(port: int, cert: str = None, key: str = None) -> subprocess.Popen:
"""Start server.py as a subprocess with the given TLS env vars."""
env = {k: v for k, v in os.environ.items()}
env["HERMES_WEBUI_HOST"] = "127.0.0.1"
env["HERMES_WEBUI_PORT"] = str(port)
env.pop("HERMES_WEBUI_TLS_CERT", None)
env.pop("HERMES_WEBUI_TLS_KEY", None)
if cert:
env["HERMES_WEBUI_TLS_CERT"] = cert
if key:
env["HERMES_WEBUI_TLS_KEY"] = key
env["HERMES_WEBUI_STATE_DIR"] = str(Path(tempfile.mkdtemp()))
proc = subprocess.Popen(
[os.sys.executable, str(ROOT / "server.py")],
env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True,
)
return proc


# ── Test class ──────────────────────────────────────────────────────────────

class TestTLSConfigFlag(unittest.TestCase):

def test_tls_enabled_true_when_both_env_set(self):
code = textwrap.dedent("""\
import os
os.environ['HERMES_WEBUI_TLS_CERT'] = '/tmp/cert.pem'
os.environ['HERMES_WEBUI_TLS_KEY'] = '/tmp/key.pem'
from api.config import TLS_ENABLED
print(TLS_ENABLED)
""")
r = subprocess.run(
[os.sys.executable, "-c", code],
capture_output=True, text=True, timeout=10,
cwd=str(ROOT),
)
self.assertEqual(r.stdout.strip(), "True")

def test_tls_enabled_false_when_env_absent(self):
env = {k: v for k, v in os.environ.items()
if k not in ("HERMES_WEBUI_TLS_CERT", "HERMES_WEBUI_TLS_KEY")}
code = textwrap.dedent("""\
import os
os.environ.pop('HERMES_WEBUI_TLS_CERT', None)
os.environ.pop('HERMES_WEBUI_TLS_KEY', None)
from api.config import TLS_ENABLED
print(TLS_ENABLED)
""")
r = subprocess.run(
[os.sys.executable, "-c", code],
capture_output=True, text=True, timeout=10,
cwd=str(ROOT), env=env,
)
self.assertEqual(r.stdout.strip(), "False")

def test_tls_enabled_false_when_only_cert_set(self):
env = {k: v for k, v in os.environ.items()
if k not in ("HERMES_WEBUI_TLS_CERT", "HERMES_WEBUI_TLS_KEY")}
env["HERMES_WEBUI_TLS_CERT"] = "/tmp/cert.pem"
code = textwrap.dedent("""\
from api.config import TLS_ENABLED
print(TLS_ENABLED)
""")
r = subprocess.run(
[os.sys.executable, "-c", code],
capture_output=True, text=True, timeout=10,
cwd=str(ROOT), env=env,
)
self.assertEqual(r.stdout.strip(), "False")


class TestTLSEndToEnd(unittest.TestCase):

@classmethod
def setUpClass(cls):
cls._tmpdir = Path(tempfile.mkdtemp())
cls._cert, cls._key = _gen_test_cert(cls._tmpdir)

@classmethod
def tearDownClass(cls):
with suppress(Exception):
import shutil
shutil.rmtree(cls._tmpdir, ignore_errors=True)

def tearDown(self):
if hasattr(self, "_proc") and self._proc.poll() is None:
self._proc.terminate()
try:
self._proc.wait(timeout=5)
except subprocess.TimeoutExpired:
self._proc.kill()

def test_https_server_responds_to_health(self):
port = _find_free_port()
self._proc = _start_server(port, cert=self._cert, key=self._key)
self.assertTrue(
_wait_for_server("127.0.0.1", port, use_ssl=True),
"TLS server did not start in time",
)
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
conn = http.client.HTTPSConnection("127.0.0.1", port, timeout=5, context=ctx)
conn.request("GET", "/health")
resp = conn.getresponse()
self.assertEqual(resp.status, 200)
data = json.loads(resp.read())
self.assertEqual(data.get("status"), "ok")
conn.close()

def test_http_without_tls_still_works(self):
port = _find_free_port()
self._proc = _start_server(port)
self.assertTrue(
_wait_for_server("127.0.0.1", port, use_ssl=False),
)
conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
conn.request("GET", "/health")
resp = conn.getresponse()
self.assertEqual(resp.status, 200)
data = json.loads(resp.read())
self.assertEqual(data.get("status"), "ok")
conn.close()

def test_tls_startup_failure_fallback_to_http(self):
"""Bad cert paths should print a warning and start HTTP anyway."""
port = _find_free_port()
self._proc = _start_server(
port, cert="/nonexistent/cert.pem", key="/nonexistent/key.pem",
)
# Server should be reachable over plain HTTP even though TLS setup failed
self.assertTrue(
_wait_for_server("127.0.0.1", port, use_ssl=False),
"HTTP fallback server did not start after TLS failure",
)
# Confirm TLS warning was printed
import fcntl
os.set_blocking(self._proc.stdout.fileno(), False)
output = ""
try:
output = self._proc.stdout.read(2000) or ""
except BlockingIOError:
output = ""
self.assertIn("TLS setup failed", output)


if __name__ == "__main__":
unittest.main()