NorthSec 2025 Writeup - LED Display Controller

import argparse
import asyncio
import configparser
import contextlib
import os
import logging
import random
import sys

import aiofiles
from aiohttp import web
from jinja2 import Template

logger = logging.getLogger(__name__)
routes = web.RouteTableDef()

"""
To run this app use: adev runserver led_display.py --port 1338 -- --filepath message.txt
You need to install aiohttp-devtools first
"""

weather_task = web.AppKey("weather_task", asyncio.Task[None])
message_task = web.AppKey("message_task", asyncio.Task[None])
message_data = web.AppKey("message_data", dict[str])
message_file_path = web.AppKey("message_file_path", str)
message_polling_interval = web.AppKey("message_polling_interval", int)
message = web.AppKey("message", str)
clients = web.AppKey("clients", set[web.WebSocketResponse])


async def fetch_message(file_path: str) -> str:
    if not os.path.exists(file_path):
        return "Message not found"
    async with aiofiles.open(file_path, "r", encoding="utf-8") as f:
        return (await f.read()).strip()


@routes.get("/")
async def index(request):
    return web.FileResponse(path=os.path.join("static", "index.html"))


@routes.get("/ws")
async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    request.app[clients].add(ws)
    try:
        await ws.send_str(request.app[message])
        async for msg in ws:
            if msg.type == web.WSMsgType.TEXT:
                pass
            elif msg.type == web.WSMsgType.CLOSED:
                break
            elif msg.type == web.WSMsgType.ERROR:
                logger.error("WebSocket error:", ws.exception())
    finally:
        request.app[clients].remove(ws)
    return ws


async def background_tasks(app):

    async def broadcast(message: str):
        for user in app[clients]:
            await user.send_str(message)

    async def message_watcher():
        try:
            while True:
                await asyncio.sleep(app[message_polling_interval])
                try:
                    message_template = await fetch_message(app[message_file_path])
                    new_message = await asyncio.wait_for(asyncio.to_thread(Template(message_template).render, app[message_data]),timeout=30)
                except Exception as e:
                    logger.exception(e)
                    continue
                if new_message != app[message]:
                    app[message] = new_message
                    await broadcast(new_message)
        except Exception as e:
            logger.exception(e)
            raise

    app[message_data].update(
        {
            "temperature": 15,
            "wind_speed": 0,
            "condition": "sunny",
        }
    )

    async def update_weather(polling=300):
        conditions = ["sunny", "cloudy", "rainy", "stormy", "foggy", "windy"]
        try:
            while True:
                app[message_data]["temperature"] = max(min(app[message_data]["temperature"] + random.uniform(-1, 1), 25), 5)
                app[message_data]["condition"] = random.choice(conditions)
                match app[message_data]["condition"]:
                    case "stormy":
                        app[message_data]["wind_speed"] = random.uniform(50, 100)
                    case "windy":
                        app[message_data]["wind_speed"] = random.uniform(30, 50)
                    case _:
                        app[message_data]["wind_speed"] = random.uniform(0, 30)
                await asyncio.sleep(polling)
        except Exception as e:
            logger.exception(e)
            raise

    app[message_task] = asyncio.create_task(message_watcher())
    app[weather_task] = asyncio.create_task(update_weather())

    yield

    app[message_task].cancel()
    app[weather_task].cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await app[message_task]
        await app[weather_task]


async def create_app():
    app = web.Application()

    logging.basicConfig(
        level=logging.DEBUG,
        format="%(message)s",
        handlers=[
            logging.StreamHandler(sys.stdout),
        ],
    )
    logging.getLogger("aiohttp").setLevel(logging.ERROR)

    parser = argparse.ArgumentParser(description="LED display")
    parser.add_argument("--file-path", required=True, help="file path of the message to display")
    parser.add_argument(
        "--polling-interval",
        default=2,
        help="polling time to check the message file",
        type=int,
    )
    args = parser.parse_args()

    app[message] = ""
    app[message_file_path] = args.file_path
    app[message_polling_interval] = args.polling_interval
    app[message_data] = {}
    app[clients] = set()

    config = configparser.ConfigParser()
    config.read("config.ini")
    app[message_data].update(config["message_data"])
    app.add_routes(routes)
    app.add_routes([web.static("/", "./static")])
    app.cleanup_ctx.append(background_tasks)

    return app


if __name__ == "__main__":
    app = create_app()
    web.run_app(app, host="::", port=8080)

#!/usr/bin/env python
import re
import os
import html
import base64
import pexpect

from ipaddress import ip_address
from flask import Flask, request, url_for

app = Flask(__name__)

DEFAULT_HOST = "led-display.ctf"
DEFAULT_PORT = "22"
DEFAULT_USERNAME = "service"
DEFAULT_PASSWORD = "Bonsecours123!"
FUNCTIONS = ["list", "change_template", "custom_message"]

def do_ssh(host, port, username, function, argument):
    ret = ""
    ssh_config_file = "/tmp/ssh_0c69c288d63d"
    command = "ssh"
    args = ["led-display"]

    config = open(ssh_config_file, "w")
    config.write(
        "Host led-display\n"
        "CheckHostIP no\n"
        "StrictHostKeyChecking no\n"
        "ConnectTimeout 30\n"
        "IdentityFile controller_key\n"
        "HostName %s\n"
        "Port %s\n"
        "User %s\n"
        "UserKnownHostsFile /dev/null\n"
        "GlobalKnownHostsFile /dev/null\n"
        "PreferredAuthentications publickey\n" %
        (host, port, username))
    config.close()

    allargs = [ "-F", config.name ]
    allargs.extend(args)
    if function in FUNCTIONS:
        allargs.append("--")
        if function == FUNCTIONS[0]:
            allargs.append("-l")
        elif function == FUNCTIONS[1]:
            allargs.append(f"-t{argument}")
        elif function == FUNCTIONS[2]:
            allargs.append(f"-m{argument}")

    try:
        ssh = pexpect.spawn(command, allargs, timeout=5, encoding="UTF-8")

        ssh.expect(pexpect.EOF)
        ret = ssh.before
    except pexpect.EOF:
        ssh_error = ssh.before.strip().split("\n")[-1]
        if ssh_error.endswith("Connection timed out"):
            raise Exception("SSH connection timed out: %s" % ssh_error)
        elif ssh_error.startswith("Permission denied"):
            raise Exception("Login failure: %s" % ssh_error)
        elif ssh_error.startswith("ssh: connect to host"):
            raise Exception("SSH connection error: %s" % ssh_error)
        else:
            raise Exception(ssh.before.strip())
    except pexpect.TIMEOUT:
        raise Exception("SSH command timed out: %s" % ssh.before.strip())
    except pexpect.ExceptionPexpect as e:
        raise Exception(str(e))
    finally:
        os.remove(ssh_config_file)

    # Check for invalid path.
    if ret.strip().endswith("No such file or directory"):
        raise Exception("SSH error: %s" % ret.strip())

    ssh.close()

    return ret

@app.route("/", methods=["GET"])
def index():
    ret = ""
    errors = []

    host = request.args.get("host", DEFAULT_HOST)
    port = request.args.get("port", DEFAULT_PORT)
    username = request.args.get("username", DEFAULT_USERNAME)
    password = request.args.get("password", "")
    function = request.args.get("function", "")
    argument = request.args.get("argument", "")

    if request.args.get("host", None) or request.args.get("port", None) or request.args.get("username", None) or request.args.get("function", None):
        if not function or not function in FUNCTIONS:
            errors.append(f"Invalid parameter: function. Must be: {', '.join(FUNCTIONS)}")
        elif not argument and function in FUNCTIONS[1:2]:
            errors.append("Must not be empty: argument.")

        try:
            ip_address(host)
        except:
            if not re.match(r"^[a-zA-Z0-9\-][a-z0-9A-Z\.\-]+[a-zA-Z]$", host):
                errors.append("Invalid parameter: host.")

        if(not port.isnumeric() or int(port) < 1 or int(port) > 65538):
            errors.append("Invalid parameter: port.")

        if not username:
            errors.append("Must not be empty: username.")

        if not errors:
            try:
                ret = do_ssh(host=host, port=port, username=username, function=function, argument=argument)
            except Exception as e:
                ret = str(e)
        else:
            ret = "\n".join(errors)

    return f"""<html>
    <head>
        <title>LED Display Controller</title>
    </head>
    <style>
    *
    {{
        color: green;
        font-family: "Courier New", Courier, monospace;
    }}

...
</html>
"""