client2server/package/src/client2server-unified.lua

--[[
    client2server-unified.lua - Bidirectional event forwarder for OpenWrt
    Copyright (c) 2026 Luis Rosales - MIT License
    
    Uses coroutines for concurrent event monitoring + command execution
]]

-- ============================================================================
-- CONFIG
-- ============================================================================

local cfg = {
    -- General
    enabled = true,
    check_interval = 5,
    log_level = "info",
    buffer_file = "/tmp/event_buffer",
    max_buffer = 100,
    -- Server
    server_url = os.getenv("SERVER_URL") or "wss://your-server.com:3843",
    server_token = os.getenv("SERVER_TOKEN") or "secret-token",
    reconnect_delay = 5,
    ping_interval = 30,
    timeout = 10,
    -- Router
    router_id = os.getenv("ROUTER_ID") or "",
    hostname = "",
    -- WAN
    wan_interface = "wan",
    wan_device = "eth0",
    monitor_dhcp = true,
    monitor_wan = true,
    monitor_ip_change = true,
    -- Events
    send_dhcp = true,
    send_wan = true,
    send_uptime = false,
    send_version = true,
}

-- Load UCI config (all settings)
pcall(function()
    local uci = require("luci.model.uci").cursor()
    -- General
    cfg.enabled = uci:get("client2server", "general", "enabled") or cfg.enabled
    cfg.check_interval = tonumber(uci:get("client2server", "general", "check_interval")) or cfg.check_interval
    cfg.log_level = uci:get("client2server", "general", "log_level") or cfg.log_level
    cfg.buffer_file = uci:get("client2server", "general", "buffer_file") or cfg.buffer_file
    cfg.max_buffer = tonumber(uci:get("client2server", "general", "max_buffer")) or cfg.max_buffer
    -- Server
    cfg.server_url = uci:get("client2server", "server", "url") or cfg.server_url
    cfg.server_token = uci:get("client2server", "server", "token") or cfg.server_token
    cfg.reconnect_delay = tonumber(uci:get("client2server", "server", "reconnect_delay")) or cfg.reconnect_delay
    cfg.ping_interval = tonumber(uci:get("client2server", "server", "ping_interval")) or cfg.ping_interval
    cfg.timeout = tonumber(uci:get("client2server", "server", "timeout")) or cfg.timeout
    -- Router
    cfg.router_id = uci:get("client2server", "router", "id") or cfg.router_id
    cfg.hostname = uci:get("client2server", "router", "hostname") or cfg.hostname
    -- WAN
    cfg.wan_interface = uci:get("client2server", "wan", "interface") or cfg.wan_interface
    cfg.wan_device = uci:get("client2server", "wan", "device") or cfg.wan_device
    cfg.monitor_dhcp = uci:get("client2server", "wan", "monitor_dhcp") or cfg.monitor_dhcp
    cfg.monitor_wan = uci:get("client2server", "wan", "monitor_wan") or cfg.monitor_wan
    cfg.monitor_ip_change = uci:get("client2server", "wan", "monitor_ip_change") or cfg.monitor_ip_change
    -- Events
    cfg.send_dhcp = uci:get("client2server", "events", "send_dhcp") or cfg.send_dhcp
    cfg.send_wan = uci:get("client2server", "events", "send_wan") or cfg.send_wan
    cfg.send_uptime = uci:get("client2server", "events", "send_uptime") or cfg.send_uptime
    cfg.send_version = uci:get("client2server", "events", "send_version") or cfg.send_version
end)

if cfg.router_id == "" then
    local f = io.popen("hostname")
    cfg.router_id = f and (f:read("*a") or ""):gsub("%s+$", "") or "unknown"
    if f then f:close() end
end

-- ============================================================================
-- LOGGING
-- ============================================================================

local function log(level, msg)
    os.execute(string.format('logger -t "client2server" -p user.%s "%s"', level, msg:gsub('"', '\\"')))
end

local function log_info(msg)  log("info", msg) end
local function log_err(msg)  log("err", msg) end

-- ============================================================================
-- JSON
-- ============================================================================

local function json_encode(t)
    local parts = {}
    for k, v in pairs(t) do
        if type(v) == "string" then
            table.insert(parts, string.format('"%s": "%s"', k, v:gsub('"', '\\"')))
        elseif type(v) == "number" then
            table.insert(parts, string.format('"%s": %s', k, tostring(v)))
        elseif type(v) == "boolean" then
            table.insert(parts, string.format('"%s": %s', k, tostring(v)))
        elseif type(v) == "table" then
            table.insert(parts, string.format('"%s": %s', k, json_encode(v)))
        end
    end
    return "{" .. table.concat(parts, ",") .. "}"
end

local function json_decode(str)
    local result = {}
    for k, v in str:gmatch('"([^"]+)":%s*"([^"]*)"') do result[k] = v end
    for k, v in str:gmatch('"([^"]+)":%s*(%d+)') do result[k] = tonumber(v) end
    for k, v in str:gmatch('"([^"]+)":%s*(%a+)') do result[k] = v end
    return result
end

-- ============================================================================
-- BUFFER
-- ============================================================================

local buffer = { events = {}, dirty = false }

function buffer.init()
    local f = io.open(cfg.buffer_file, "r")
    if f then
        for line in f:lines() do
            if line and line ~= "" then
                table.insert(buffer.events, line)
            end
        end
        f:close()
    end
end

function buffer.save()
    if not buffer.dirty then return end
    local f = io.open(cfg.buffer_file, "w")
    if not f then return end
    for _, ev in ipairs(buffer.events) do f:write(ev .. "\n") end
    f:close()
    buffer.dirty = false
end

function buffer.add(json_event)
    -- Warn if buffer is getting full
    if #buffer.events > cfg.max_buffer * 0.8 then
        log_err("Buffer almost full: " .. #buffer.events .. "/" .. cfg.max_buffer)
    end
    
    table.insert(buffer.events, json_event)
    if #buffer.events > cfg.max_buffer then
        -- Buffer full - drop oldest to make room
        table.remove(buffer.events, 1)
        log_err("Buffer overflow: dropping oldest event")
    end
    buffer.dirty = true
end

function buffer.flush(send_fn)
    if #buffer.events == 0 then return end
    local i = 1
    while i <= #buffer.events do
        if send_fn(buffer.events[i]) then
            table.remove(buffer.events, i)
            buffer.dirty = true
        else i = i + 1 end
    end
    buffer.save()
end

-- ============================================================================
-- WEBSOCKET (RFC 6455 - Proper WebSocket)
-- ============================================================================

local ws = { sock = nil, connected = false, key = "" }

-- Simple base64 encoder
local function base64_encode(data)
    local b64 = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
    local result = {}
    local i = 1
    while i <= #data do
        local b1, b2, b3 = string.byte(data, i, i+2)
        b2 = b2 or 0
        b3 = b3 or 0
        table.insert(result, string.sub(b64, math.floor(b1/4)+1, math.floor(b1/4)+1))
        table.insert(result, string.sub(b64, ((b1%16)*4) + math.floor(b2/16)+1, ((b1%16)*4) + math.floor(b2/16)+1))
        if i+1 > #data then table.insert(result, "=") else
            table.insert(result, string.sub(b64, ((b2%16)*4) + math.floor(b3/64)+1, ((b2%16)*4) + math.floor(b3/64)+1))
        end
        if i+2 > #data then table.insert(result, "=") else
            table.insert(result, string.sub(b64, (b3%64)+1, (b3%64)+1))
        end
        i = i + 3
    end
    return table.concat(result)
end

-- SHA1 (for WebSocket handshake)
local function sha1_binary(data)
    local f = io.popen("echo -n '" .. data:gsub("'", "'\\''") .. "' | openssl sha1 -binary | base64 | tr -d '\\n' 2>/dev/null")
    if f then
        local result = f:read("*a")
        f:close()
        return result:gsub("%s+$", "")
    end
    return ""
end

-- Compute Sec-WebSocket-Accept
local function compute_accept(key)
    local combined = key .. "258EAFA5-E914-47DA-95CA-C5C753455362"
    return sha1_binary(combined)
end

function ws.connect(url)
    local is_ssl = url:match("wss://") ~= nil
    local host = url:match("wss?://([^:/]+)")
    local port = url:match(":(%d+)") or (is_ssl and "443" or "80")
    if not host then return nil end

    local sock = require("socket").tcp()
    sock:settimeout(10)

    local ok, err = sock:connect(host, tonumber(port))
    if not ok then
        log_err("Cannot connect to " .. host .. ":" .. port .. ": " .. tostring(err))
        return nil
    end

    -- Generate random Sec-WebSocket-Key
    local key = ""
    for i = 1, 16 do key = key .. string.char(math.random(32, 126)) end
    key = base64_encode(key)
    ws.key = key

    local request = "GET /ws HTTP/1.1\r\n" ..
            "Host: " .. host .. ":" .. port .. "\r\n" ..
            "Upgrade: websocket\r\n" ..
            "Connection: Upgrade\r\n" ..
            "Sec-WebSocket-Key: " .. key .. "\r\n" ..
            "Sec-WebSocket-Version: 13\r\n" ..
            "Origin: http://" .. host .. "\r\n" ..
            "\r\n"

    sock:send(request)


    -- Read response
    local response = {}
    sock:settimeout(5)
    for i = 1, 20 do
        local line = sock:receive("*l")
        if not line or line == "" then break end
        table.insert(response, line)
    end

    -- Check for 101 Switching Protocols
    local ok_response = false
    for _, line in ipairs(response) do
        if line:match("^HTTP/.* 101") then ok_response = true end
    end

    if not ok_response then
        log_err("WebSocket handshake failed")
        sock:close()
        return nil
    end

    ws.sock = sock
    ws.connected = true
    log_info("WebSocket connected to " .. host .. ":" .. port)
    return sock
end

function ws.send(data)
    if not ws.connected then buffer.add(data); return false end

    -- WebSocket frame: FIN(1) + opcode(1) = 0x81 (text)
    local payload = data
    local frame = string.char(0x81) .. payload

    if not pcall(function() ws.sock:send(frame) end) then
        ws.connected = false
        buffer.add(data)
        return false
    end
    return true
end

function ws.recv()
    if not ws.connected then return nil end

    ws.sock:settimeout(0.5)
    local data, err = ws.sock:receive("*l")
    ws.sock:settimeout(10)

    if err and err ~= "timeout" then
        ws.connected = false
        return nil
    end

    -- Strip WebSocket frame header (first byte)
    if data and #data > 1 then
        data = data:sub(2)
    end

    return data
end

function ws.close()
    if ws.sock and ws.connected then
        pcall(function() ws.sock:send(string.char(0x88, 0x00)) end)
    end
    if ws.sock then pcall(ws.sock.close, ws.sock) end
    ws.sock = nil
    ws.connected = false
end

-- ============================================================================
-- COMMAND EXECUTOR
-- ============================================================================

function execute_command(cmd_obj)
    local cmd = cmd_obj.command or ""
    local args = cmd_obj.args or {}
    
    log_info("Executing: " .. cmd)
    
    local result = { success = false, output = "", error = "" }
    
    if cmd == "uci_set" then
        local config = args.config
        local section = args.section
        local option = args.option
        local value = args.value
        
        if config and section and option and value then
            local c = string.format("uci set %s.%s.%s='%s'", config, section, option, value)
            local f = io.popen(c)
            result.output = f and f:read("*a") or ""
            if f then f:close() end
            os.execute("uci commit " .. config)
            result.success = true
            log_info("UCI set: " .. config .. "." .. section .. "." .. option .. " = " .. value)
        else
            result.error = "Missing params"
        end
        
    elseif cmd == "shell" then
        local shell_cmd = args.command
        if shell_cmd then
            local f = io.popen(shell_cmd)
            result.output = f and f:read("*a") or ""
            if f then f:close() end
            result.success = true
        else
            result.error = "No command"
        end
        
    elseif cmd == "reboot" then
        os.execute("sync && reboot &")
        result.success = true
        result.output = "Reboot scheduled"
        
    elseif cmd == "wifi_restart" then
        os.execute("/etc/init.d/network restart")
        os.execute("/etc/init.d/wireless restart")
        result.success = true
        
    elseif cmd == "status" then
        local f = io.popen("ubus call network getStatus")
        result.output = f and f:read("*a") or "{}"
        if f then f:close() end
        result.success = true
        
    else
        result.error = "Unknown: " .. cmd
    end
    
    return result
end

-- ============================================================================
-- COROUTINES
-- ============================================================================

-- Monitor DHCP
local function co_dhcp()
    local leases = {}
    
    while true do
        local f = io.open("/var/lib/dnsmasq/dnsmasq.leases", "r")
        if f then
            local current = {}
            for line in f:lines() do
                local ts, mac, ip, name = line:match("(%d+)%s+(%S+)%s+(%S+)%s+(%S+)")
                if mac then
                    current[mac] = { ip = ip, hostname = name }
                    if not leases[mac] then
                        local ev = json_encode({
                            router_id = cfg.router_id, hostname = cfg.router_id,
                            event_type = "dhcp_lease_new",
                            payload = { mac = mac, ip = ip, hostname = name }
                        })
                        ws.send(ev)
                        log_info("DHCP: " .. mac .. " -> " .. ip)
                    end
                end
            end
            f:close()
            
            for mac in pairs(leases) do
                if not current[mac] then
                    local ev = json_encode({
                        router_id = cfg.router_id, hostname = cfg.router_id,
                        event_type = "dhcp_lease_expire",
                        payload = { mac = mac, old_ip = leases[mac].ip }
                    })
                    ws.send(ev)
                    log_info("DHCP expire: " .. mac)
                end
            end
            leases = current
        end
        coroutine.yield(cfg.check_interval)
    end
end

-- Monitor WAN
local function co_wan()
    local link_last, ip_last = nil, nil
    
    local f = io.open("/sys/class/net/" .. cfg.wan_device .. "/carrier", "r")
    if f then link_last = (f:read("*a") or "":find("^1") == 1; f:close() end
    
    while true do
        -- Link
        f = io.open("/sys/class/net/" .. cfg.wan_device .. "/carrier", "r")
        local link_now = f and ((f:read("*a") or ""):find("^1") == 1) or false
        if f then f:close() end
        
        if link_now ~= link_last then
            link_last = link_now
            local ev = json_encode({
                router_id = cfg.router_id, hostname = cfg.router_id,
                event_type = link_now and "wan_link_up" or "wan_link_down",
                payload = { device = cfg.wan_device }
            })
            ws.send(ev)
            log_info("WAN link: " .. (link_now and "up" or "down"))
        end
        
        -- IP
        f = io.popen("ubus call network.interface." .. cfg.wan_interface .. " status 2>/dev/null")
        local ip_now = nil
        if f then
            local st = f:read("*a")
            f:close()
            ip_now = st and st:match('"address"%s*:%s*"([^"]+)"')
        end
        
        if ip_now and ip_now ~= ip_last then
            local ev = json_encode({
                router_id = cfg.router_id, hostname = cfg.router_id,
                event_type = ip_last and "wan_dhcp_changed" or "wan_dhcp_new",
                payload = { old_ip = ip_last, new_ip = ip_now }
            })
            ws.send(ev)
            log_info("WAN IP: " .. (ip_last or "none") .. " -> " .. ip_now)
            ip_last = ip_now
        end
        
        coroutine.yield(cfg.check_interval)
    end
end

-- 🔥 COROUTINE: Command Listener (handles server → router commands)
local function co_commands()
    while true do
        if ws.connected then
            -- Check for incoming data
            local data = ws.recv()
            
            if data and data:match("^{") then
                log_info("Received command: " .. data:sub(1, 100))
                
                -- Parse JSON command
                local cmd_obj = json_decode(data)
                
                if cmd_obj.command then
                    -- Execute command
                    local result = execute_command(cmd_obj)
                    
                    -- Send result back
                    local resp = json_encode({
                        router_id = cfg.router_id,
                        event_type = "command_result",
                        payload = {
                            command = cmd_obj.command,
                            command_id = cmd_obj.id or "",
                            success = result.success,
                            output = result.output,
                            error = result.error
                        }
                    })
                    ws.send(resp)
                    
                    log_info("Command done: " .. cmd_obj.command .. " = " .. (result.success and "OK" or result.error))
                end
            end
        end
        
        coroutine.yield(1)  -- Check every second for commands
    end
end

-- COROUTINE: Auto-reconnect (keeps connection alive)
local function co_connect()
    local retry_delay = 30  -- Start at 30s
    local max_delay = 300    -- Max 5 minutes
    
    while true do
        -- Urgency based on buffer state
        local buffer_fullness = #buffer.events / cfg.max_buffer
        
        if buffer_fullness > 0.8 then
            log_err("Buffer critical at " .. string.format("%.0f%%", buffer_fullness * 100) .. " - aggressive reconnect")
            retry_delay = 10  -- Aggressive when buffer filling
        end
        
        if not ws.connected then
            log_info("Connecting to " .. cfg.server_url .. "...")
            
            local sock = ws.connect(cfg.server_url)
            
            if sock then
                ws.connected = true
                retry_delay = 30  -- Reset on success
                log_info("Connected!")
                buffer.flush(function(d) return ws.send(d) end)
            else
                log_err("Connection failed, retry in " .. retry_delay .. "s")
                coroutine.yield(retry_delay)
                retry_delay = math.min(retry_delay * 2, max_delay)
            end
        else
            -- Even when connected, periodically flush to clear buffer buildup
            buffer.flush(function(d) return ws.send(d) end)
            coroutine.yield(30)
        end
    end
end

-- ============================================================================
-- MAIN
-- ============================================================================

local function scheduler()
    local cos = {
        co_dhcp = coroutine.create(co_dhcp),
        co_wan = coroutine.create(co_wan),
        co_commands = coroutine.create(co_commands),
        co_connect = coroutine.create(co_connect),  -- Auto-reconnect!
    }
    
    while true do
        -- Resume all coroutines
        for name, co in pairs(cos) do
            if coroutine.status(co) == "suspended" then
                local _, delay = coroutine.resume(co)
                -- Small stagger to avoid spikes
                if delay then os.execute("sleep 0.1") end
            end
        end
        
        -- Flush buffer when connected
        if ws.connected then
            buffer.flush(function(d) return ws.send(d) end)
        end
        
        os.execute("sleep 1")
    end
end

local function main()
    log_info("Starting client2server...")
    log_info("Router: " .. cfg.router_id)
    log_info("Server: " .. cfg.server_url)
    
    buffer.init()
    
    local pf = io.open(cfg.pid_file, "w")
    if pf then pf:write(tostring(os.getpid())); pf:close() end
    
    -- Connect
    log_info("Initial connection...")
    local sock = ws.connect(cfg.server_url)
    
    if sock then
        ws.connected = true
        log_info("Connected!")
        buffer.flush(function(d) return ws.send(d) end)
    else
        log_err("Connection failed")
    end
    
    scheduler()
end

main()