malp/cgi-bin/malp.rb

#!/usr/bin/env ruby

require "cgi"
require "erb"
require "digest"
require "json"
require "securerandom"
require "socket"

ASSETS_DIR = File.join(__dir__, "../assets")
DATA_DIR = File.join(__dir__, "../data")
SESSIONS_FILE = File.join(DATA_DIR, "sessions.txt")
USER_FILE = File.join(DATA_DIR, "user.txt")

cgi = CGI.new

hostname = File.read("/etc/hostname").strip rescue "localhost"

def load_sessions
  return [] unless File.exist?(SESSIONS_FILE)
  now = Time.now.to_i
  max_age = 3 * 7 * 24 * 60 * 60 # 3 weeks
  sessions = File.readlines(SESSIONS_FILE).filter_map do |line|
    token, timestamp = line.strip.split(":", 2)
    next if token.nil? || token.empty?
    [token, timestamp.to_i]
  end
  active, expired = sessions.partition { |_, ts| now - ts < max_age }
  if expired.any?
    File.write(SESSIONS_FILE, active.map { |t, ts| "#{t}:#{ts}" }.join("\n") + "\n")
  end
  active
end

def valid_session?(token)
  return false if token.nil? || token.empty?
  load_sessions.any? { |t, _| t == token }
end

def check_credentials(username, password)
  return false unless File.exist?(USER_FILE)
  stored_user, salt, stored_hash_hex2 = File.read(USER_FILE).strip.split(":", 3)
  return false unless username == stored_user
  stored_hash = [stored_hash_hex2].pack("H*")
  computed_hash = Digest::SHA256.hexdigest(salt + password)
  stored_hash == computed_hash
end

def create_session
  token = SecureRandom.hex(32)
  File.open(SESSIONS_FILE, "a") { |f| f.puts("#{token}:#{Time.now.to_i}") }
  token
end

def remove_session(token)
  return if token.nil? || token.empty?
  sessions = load_sessions.reject { |t, _| t == token }
  File.write(SESSIONS_FILE, sessions.map { |t, ts| "#{t}:#{ts}" }.join("\n") + "\n")
end

session_token = (cgi.cookies["MALP"] || []).first
authenticated = valid_session?(session_token)
cookie = nil

if cgi.request_method == "POST" && authenticated && cgi.params["action"]&.first == "logout"
  remove_session(session_token)
  cookie = CGI::Cookie.new("name" => "MALP", "value" => "", "path" => "/", "expires" => Time.at(0))
  print cgi.header("Status" => "303 See Other", "Location" => ENV["REQUEST_URI"], "cookie" => cookie)
  exit
end

if cgi.request_method == "POST" && !authenticated
  username = cgi.params["username"]&.first.to_s
  password = cgi.params["password"]&.first.to_s

  if check_credentials(username, password)
    token = create_session
    cookie = CGI::Cookie.new("name" => "MALP", "value" => token, "path" => "/")
    print cgi.header("Status" => "303 See Other", "Location" => ENV["REQUEST_URI"], "cookie" => cookie)
    exit
  end
end

if cgi.params.key?("content")
  if authenticated
    html = ""

    begin
      sock = UNIXSocket.new("/run/malpd/malpd.sock")
      sock.puts "info"
      info = JSON.parse(sock.gets)
      sock.close
    rescue
      cgi.out("type" => "text/html", "charset" => "UTF-8") do
        "Could not connect to malpd socket"
      end
      exit
    end

    thousands = ->(n) { n.to_s.gsub(/(\d)(?=(\d{3})+$)/, '\1,') }

    drive_stats = lambda do |dt, smart|
      stats = []

      case smart["passed"]
      when true  then stats << ["ok",  "SMART PASSED"]
      when false then stats << ["bad", "SMART FAILED"]
      end

      if (t = smart["temperature"])
        cls = t >= 60 ? "bad" : t >= 50 ? "warn" : "ok"
        stats << [cls, "🌡 #{t} °C"]
      end

      if (poh = smart["power_on_hours"])
        stats << ["ok", "POH #{thousands.(poh)} h"]
      end

      if dt == "nvme"
        if (p = smart["percentage_used"])
          cls = p >= 90 ? "bad" : p >= 70 ? "warn" : "ok"
          stats << [cls, "Wear #{p}%"]
        end
        if (m = smart["media_errors"])
          stats << [m > 0 ? "bad" : "ok", "Media err #{m}"]
        end
        if (cw = smart["critical_warning"]) && cw != 0
          stats << ["bad", "Crit warn 0x#{cw.to_s(16)}"]
        end
      else
        if (r = smart["reallocated_sectors"])
          stats << [r > 0 ? "warn" : "ok", "Realloc #{r}"]
        end
        if (s = smart["spin_retry"])
          stats << [s > 0 ? "warn" : "ok", "Spin retry #{s}"]
        end
        if (p = smart["pending_sectors"])
          stats << [p > 0 ? "warn" : "ok", "Pending #{p}"]
        end
        if (c = smart["crc_errors"])
          stats << [c > 0 ? "warn" : "ok", "CRC #{c}"]
        end
      end

      stats
    end

    info.each do |entry|
      case entry["type"]
      when "drive"
        dt = entry["drivetype"]
        name = CGI.escapeHTML(entry["name"].to_s)
        model = CGI.escapeHTML(entry["model"])
        capacity = CGI.escapeHTML(entry["capacity"])
        smart = entry["smart"]

        stats = smart ? drive_stats.(dt, smart) : []
        card_cls = if stats.any? { |c, _| c == "bad" } then "bad"
                   elsif stats.any? { |c, _| c == "warn" } then "warn"
                   else "ok"
                   end

        stats_html = stats.map { |c, txt|
          %(<span class="stat #{c}">#{CGI.escapeHTML(txt)}</span>)
        }.join
        stats_block = stats.empty? ? "" :
          %(<div class="drive-stats" style="margin-top:0.6rem;">#{stats_html}</div>)

        title = name.empty? ? model : "#{name} · #{model}"

        html << <<~HTML
          <div class="card #{card_cls} span-2">
            <div class="card-header">
              <span class="card-title">#{title} <span class="card-capacity">#{capacity}</span> <span class="drive-type #{dt}">#{dt.upcase}</span></span>
            </div>
            #{stats_block}
          </div>
        HTML
      end
    end

    cgi.out("type" => "text/html", "charset" => "UTF-8") do
      html
    end
  else
    cgi.out("Status" => "403 Forbidden", "type" => "text/plain", "charset" => "UTF-8") do
      "Forbidden"
    end
  end
  exit
end

template = ERB.new(File.read(File.join(ASSETS_DIR, "page.erb")))

out_params = { "type" => "text/html", "charset" => "UTF-8" }
out_params["cookie"] = cookie if cookie

cgi.out(out_params) do
  template.result(binding)
end