Deploy frame via agent (#142)

Author: mariusandra

Dockerfile

@@ -61,13 +61,21 @@ RUN apt-get remove -y nodejs curl build-essential libffi-dev ca-certificates gnu
     && rm -rf /app/frontend/node_modules \
     && rm -rf /var/lib/apt/lists/* /root/.npm
 
-# Change back to the main directory
+# Install frameos nim deps
 WORKDIR /app/frameos
 
 COPY frameos/frameos.nimble ./
 COPY frameos/nimble.lock ./
 COPY frameos/nim.cfg ./
 
+RUN nimble install -d -y && nimble setup
+
+# Install frameos agent nim deps
+WORKDIR /app/agent
+
+COPY agent/frameos_agent.nimble ./
+COPY agent/nimble.lock ./
+
 # Cache nimble deps for when deploying on frame
 RUN nimble install -d -y && nimble setup
 

agent/frameos_agent.service

@@ -14,6 +14,7 @@ RestartSec=5
 LimitNOFILE=65536
 PrivateTmp=yes
 ProtectSystem=full
+ReadWritePaths=/etc/systemd/system /etc/cron.d /boot /boot/firmware
 
 [Install]
 WantedBy=multi-user.target

agent/src/frameos_agent.nim

@@ -1,5 +1,5 @@
 import std/[algorithm, segfaults, strformat, strutils, asyncdispatch, terminal,
-            times, os, sysrand, httpclient, osproc, streams, unicode]
+            times, os, sysrand, httpclient, osproc, streams, unicode, typedthreads]
 import checksums/md5
 import json, jsony
 import ws
@@ -52,6 +52,14 @@ type
     network*: NetworkConfig
     agent*: AgentConfig
 
+  WhichStream* = enum stdoutStr, stderrStr
+
+  LineMsg* = object
+    stream*: WhichStream ## stdout or stderr
+    txt*: string         ## one complete line
+    done*: bool          ## true when process finished
+    exit*: int           ## exit code (only valid if done)
+
 # ----------------------------------------------------------------------------
 # Config IO (fails hard if unreadable)
 # ----------------------------------------------------------------------------
@@ -199,7 +207,63 @@ proc recvBinary(ws: WebSocket): Future[string] {.async.} =
       discard # ignore text, pong …
 
 # ----------------------------------------------------------------------------
-# Challenge-response handshake (server-initiated)
+# Shell command helpers
+# ----------------------------------------------------------------------------
+type
+  StreamParams = tuple[
+    pipe: Stream;
+    which: WhichStream;
+    ch: ptr Channel[LineMsg]
+  ]
+
+  OutParams = tuple[
+    process: Process;
+    ch: ptr Channel[LineMsg]
+  ]
+
+proc readErr(p: StreamParams) {.thread.} =
+  let (pipe, which, ch) = p
+  var line: string
+  while pipe.readLine(line):
+    ch[].send LineMsg(stream: which, txt: line, done: false)
+
+proc readOut(p: OutParams) {.thread.} =
+  let (process, ch) = p
+  var line: string
+  while process.outputStream.readLine(line):
+    ch[].send LineMsg(stream: stdoutStr, txt: line, done: false)
+  let rc = process.waitForExit() # EOF ⇒ child quit
+  ch[].send LineMsg(done: true, exit: rc) # final signal
+
+proc execShellThreaded(rawCmd: string; ws: WebSocket;
+                       cfg: FrameConfig; id: string): Future[void] {.async.} =
+  var ch: Channel[LineMsg]
+  ch.open()
+  var p = startProcess("/bin/bash",
+          args = ["-c", rawCmd],
+          options = {poUsePath})
+
+  var thrOut: Thread[OutParams]
+  var thrErr: Thread[StreamParams]
+
+  createThread(thrOut, readOut, (p, addr ch)) # stdout + exit code
+  createThread(thrErr, readErr, (p.errorStream, stderrStr, addr ch))
+
+  while true:
+    let (ready, msg) = tryRecv(ch) # new one-arg form
+    if ready:
+      if msg.done:
+        await sendResp(ws, cfg, id, msg.exit == 0, %*{"exit": msg.exit})
+        break
+      else:
+        let which = if msg.stream == stdoutStr: "stdout" else: "stderr"
+        await streamChunk(ws, cfg, id, which, msg.txt & "\n")
+    else:
+      await sleepAsync(100)
+  ch.close()
+
+# ----------------------------------------------------------------------------
+# All command handlers
 # ----------------------------------------------------------------------------
 proc handleCmd(cmd: JsonNode; ws: WebSocket; cfg: FrameConfig): Future[void] {.async.} =
   let id = cmd{"id"}.getStr()
@@ -210,7 +274,7 @@ proc handleCmd(cmd: JsonNode; ws: WebSocket; cfg: FrameConfig): Future[void] {.a
 
   # No remote execution available
   if not cfg.agent.agentRunCommands:
-    if name != "version":
+    if name != "version": # only allow "version" command
       await sendResp(ws, cfg, id, false, %*{"error": "agentRunCommands disabled in config"})
       return
 
@@ -265,31 +329,9 @@ proc handleCmd(cmd: JsonNode; ws: WebSocket; cfg: FrameConfig): Future[void] {.a
               "binary": false})
     of "shell":
       if not args.hasKey("cmd"):
-        await sendResp(ws, cfg, id, false,
-                      %*{"error": "`cmd` missing"})
-        return
-
-      let cmdStr = args["cmd"].getStr
-
-      var p = startProcess(
-        "/bin/sh",             # command
-        args = ["-c", cmdStr], # argv
-        options = {poUsePath, poStdErrToStdOut}
-      )
-
-      let bufSize = 4096
-      var buf = newString(bufSize)
-
-      while true:
-        let n = p.outputStream.readData(addr buf[0], buf.len)
-        if n == 0:
-          if p.running: await sleepAsync(100) # no data yet – yield
-          else: break # process finished – exit loop
-        else:
-          await streamChunk(ws, cfg, id, "stdout", buf[0 ..< n])
-
-      let rc = p.waitForExit()
-      await sendResp(ws, cfg, id, rc == 0, %*{"exit": rc})
+        await sendResp(ws, cfg, id, false, %*{"error": "`cmd` missing"})
+      else:
+        asyncCheck execShellThreaded(args["cmd"].getStr(), ws, cfg, id)
 
     of "file_md5":
       let path = args{"path"}.getStr("")
@@ -452,14 +494,11 @@ proc doHandshake(ws: WebSocket; cfg: FrameConfig): Future[void] {.async.} =
     echo &"⚠️ handshake failed, unexpected action: {act} in {ackMsg}"
     raise newException(Exception, "Handshake failed: " & ackMsg)
 
-# ----------------------------------------------------------------------------
-# Heartbeat helper
-# ----------------------------------------------------------------------------
 proc startHeartbeat(ws: WebSocket; cfg: FrameConfig): Future[void] {.async.} =
   ## Keeps server-side idle-timeout at bay.
   try:
     while true:
-      await sleepAsync(20_000)
+      await sleepAsync(40_000)
       let env = makeSecureEnvelope(%*{"type": "heartbeat"}, cfg)
       await ws.send($env)
   except Exception: discard # will quit when ws closes / errors out
@@ -483,7 +522,7 @@ proc runAgent(cfg: FrameConfig) {.async.} =
         await doHandshake(ws, cfg) # throws on failure
         backoff = InitialBackoffSeconds # reset back-off
 
-        asyncCheck startHeartbeat(ws, cfg) # fire-and-forget
+        asyncCheck startHeartbeat(ws, cfg)
 
         # ── Main receive loop ───────────────────────────────────────────────
         while true:

backend/app/api/frames.py

@@ -707,11 +707,7 @@ async def api_frame_get_assets(
 
     ssh = await get_ssh_connection(db, redis, frame)
     try:
-        cmd = (
-            f"find {shlex.quote(assets_path)} \
-                \( -type f -o -type d \) \
-                -exec stat --printf='%F|%s|%Y|%n\n' {{}} +"
-        )
+        cmd = f"find {shlex.quote(assets_path)} -exec stat --printf='%F|%s|%Y|%n\\n' {{}} +"
         output: list[str] = []
         await exec_command(db, redis, frame, ssh, cmd, output, log_output=False)
     finally:
@@ -725,12 +721,13 @@ async def api_frame_get_assets(
         if len(parts) != 4:
             continue
         ftype, s, m, p = parts
-        assets.append({
-            "path": p.strip(),
-            "size": int(s),
-            "mtime": int(m),
-            "is_dir": ftype == "directory",
-        })
+        if p.strip() != assets_path:
+            assets.append({
+                "path": p.strip(),
+                "size": int(s),
+                "mtime": int(m),
+                "is_dir": ftype == "directory",
+            })
     assets.sort(key=lambda a: a["path"])
     return {"assets": assets}
 
@@ -745,14 +742,7 @@ async def api_frame_assets_sync(
     try:
         from app.models.assets import sync_assets
 
-        if await _use_agent(frame, redis):
-            await sync_assets(db, redis, frame, None)
-        else:
-            ssh = await get_ssh_connection(db, redis, frame)
-            try:
-                await sync_assets(db, redis, frame, ssh)
-            finally:
-                await remove_ssh_connection(db, redis, ssh, frame)
+        await sync_assets(db, redis, frame)
         return {"message": "Assets synced successfully"}
     except Exception as e:
         raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))

backend/app/models/assets.py

@@ -1,14 +1,12 @@
-import tempfile
 import uuid
 import os
-import asyncssh
+
 from sqlalchemy import LargeBinary, String, Text
 from sqlalchemy.orm import Session, mapped_column
 from app.models.frame import Frame
 from app.database import Base
 from arq import ArqRedis as Redis
-from app.utils.remote_exec import _use_agent, run_commands
-from app.utils.ssh_utils import exec_command
+from app.utils.remote_exec import _use_agent, run_commands, run_command, upload_file
 from app.models.log import new_log as log
 
 default_assets_path = "/srv/assets"
@@ -26,13 +24,13 @@ def to_dict(self):
             'size': len(self.data) if self.data else 0,
         }
 
-async def sync_assets(db: Session, redis: Redis, frame: Frame, ssh=None):
+async def sync_assets(db: Session, redis: Redis, frame: Frame):
     assets_path = frame.assets_path or default_assets_path
-    await make_asset_folders(db, redis, frame, ssh, assets_path)
+    await make_asset_folders(db, redis, frame, assets_path)
     if frame.upload_fonts != "none":
-        await upload_font_assets(db, redis, frame, ssh, assets_path)
+        await upload_font_assets(db, redis, frame, assets_path)
 
-async def make_asset_folders(db: Session, redis: Redis, frame: Frame, ssh, assets_path: str):
+async def make_asset_folders(db: Session, redis: Redis, frame: Frame, assets_path: str):
     if frame.upload_fonts != "none":
         cmd = (
             f"if [ ! -d {assets_path}/fonts ]; then "
@@ -52,23 +50,19 @@ async def make_asset_folders(db: Session, redis: Redis, frame: Frame, ssh, asset
             f"fi"
         )
 
-    if await _use_agent(frame, redis):
-        await run_commands(db, redis, frame, [cmd])
-    else:
-        await exec_command(db, redis, frame, ssh, cmd)
+    await run_commands(db, redis, frame, [cmd])
 
-async def upload_font_assets(db: Session, redis: Redis, frame: Frame, ssh, assets_path: str):
+async def upload_font_assets(db: Session, redis: Redis, frame: Frame, assets_path: str):
     if await _use_agent(frame, redis):
         from app.ws.agent_ws import assets_list_on_frame
         assets = await assets_list_on_frame(frame.id, assets_path + "/fonts")
         remote_fonts = {a["path"]: int(a.get("size", 0)) for a in assets}
     else:
         command = f"find {assets_path}/fonts -type f -exec stat --format='%s %Y %n' {{}} +"
-        output: list[str] = []
-        await exec_command(db, redis, frame, ssh, command, output, log_output=False)
-
+        status, stdout, _ = await run_command(db, redis, frame, command)
+        stdout_lines = stdout.splitlines()
         remote_fonts = {}
-        for line in output:
+        for line in stdout_lines:
             if not line:
                 continue
             size, mtime, path = line.split(' ', 2)
@@ -119,28 +113,17 @@ async def upload_font_assets(db: Session, redis: Redis, frame: Frame, ssh, asset
             await file_write_on_frame(frame.id, remote_path, font.data)
 
     else:
-        await log(db, redis, frame.id, "stdout", f"Uploading {len(fonts_to_upload) + len(custom_fonts_to_upload)} fonts")
         for local_path, remote_path in fonts_to_upload:
-            await asyncssh.scp(
-                local_path,
-                (ssh, remote_path),
-                recurse=False,
-            )
+            with open(local_path, "rb") as fh:
+                data = fh.read()
+            await upload_file(db, redis, frame, remote_path, data)
         for font, remote_path in custom_fonts_to_upload:
-            with tempfile.NamedTemporaryFile(suffix=".ttf", delete=False) as tmpf:
-                tempfile_path = tmpf.name
-                tmpf.write(font.data)
-            await asyncssh.scp(
-                tempfile_path,
-                (ssh, remote_path),
-                recurse=False,
-            )
-            os.remove(tempfile_path)
+            await upload_file(db, redis, frame, remote_path, font.data)
 
     await log(
         db,
         redis,
         frame.id,
         "stdout",
         f"Uploaded {len(fonts_to_upload) + len(custom_fonts_to_upload)} fonts",
-    )
+    )

backend/app/tasks/deploy_agent.py

@@ -19,8 +19,8 @@
     get_ssh_connection,
     exec_command,
     remove_ssh_connection,
-    exec_local_command,
 )
+from app.utils.local_exec import exec_local_command
 from .utils import find_nim_v2, find_nimbase_file