Add function to report devices activity

central_backend/Cargo.lock

@@ -632,6 +632,7 @@ dependencies = [
  "clap",
  "env_logger",
  "foreign-types-shared",
+ "fs4",
  "futures",
  "futures-util",
  "jsonwebtoken",
@@ -1021,6 +1022,16 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "fs4"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e8c6b3bd49c37d2aa3f3f2220233b29a7cd23f79d1fe70e5337d25fb390793de"
+dependencies = [
+ "rustix",
+ "windows-sys 0.52.0",
+]
+
 [[package]]
 name = "futures"
 version = "0.3.30"

central_backend/Cargo.toml

@@ -39,3 +39,4 @@ prettytable-rs = "0.10.0"
 chrono = "0.4.38"
 serde_yml = "0.0.12"
 bincode = "=2.0.0-rc.3"
+fs4 = { version = "0.9", features = ["sync"] }

central_backend/src/app_config.rs

@@ -283,6 +283,16 @@ impl AppConfig {
                 }
             ))
     }
+
+    /// Get logs directory
+    pub fn logs_dir(&self) -> PathBuf {
+        self.storage_path().join("logs")
+    }
+
+    /// Get the logs for a given day
+    pub fn log_of_day(&self, day: u64) -> PathBuf {
+        self.logs_dir().join(format!("{day}.log"))
+    }
 }
 
 #[cfg(test)]

central_backend/src/lib.rs

@@ -3,5 +3,6 @@ pub mod constants;
 pub mod crypto;
 pub mod devices;
 pub mod energy;
+pub mod logs;
 pub mod server;
 pub mod utils;

central_backend/src/logs/log_entry.rs

@@ -0,0 +1,17 @@
+use crate::devices::device::DeviceId;
+use crate::logs::severity::LogSeverity;
+
+#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
+pub struct LogEntry {
+    /// If no device is specified then the message comes from the backend
+    pub device_id: Option<DeviceId>,
+    pub time: u64,
+    pub severity: LogSeverity,
+    pub message: String,
+}
+
+impl LogEntry {
+    pub fn serialize(&self) -> anyhow::Result<String> {
+        Ok(serde_json::to_string(self)?)
+    }
+}

central_backend/src/logs/logs_manager.rs

@@ -0,0 +1,41 @@
+use crate::app_config::AppConfig;
+use crate::devices::device::DeviceId;
+use crate::logs::log_entry::LogEntry;
+use crate::logs::severity::LogSeverity;
+use crate::utils::time_utils::{curr_day_number, time_secs};
+use fs4::fs_std::FileExt;
+use std::fs::OpenOptions;
+use std::io::{Seek, SeekFrom, Write};
+
+pub fn save_log(
+    device: Option<&DeviceId>,
+    severity: LogSeverity,
+    message: String,
+) -> anyhow::Result<()> {
+    let log_path = AppConfig::get().log_of_day(curr_day_number());
+
+    let mut file = OpenOptions::new()
+        .append(true)
+        .create(true)
+        .open(&log_path)?;
+
+    file.lock_exclusive()?;
+    file.seek(SeekFrom::End(0))?;
+    file.write_all(
+        format!(
+            "{}\n",
+            (LogEntry {
+                device_id: device.cloned(),
+                time: time_secs(),
+                severity,
+                message,
+            })
+            .serialize()?
+        )
+        .as_bytes(),
+    )?;
+    file.flush()?;
+    file.unlock()?;
+
+    Ok(())
+}

central_backend/src/logs/mod.rs

@@ -0,0 +1,3 @@
+pub mod log_entry;
+pub mod logs_manager;
+pub mod severity;

central_backend/src/logs/severity.rs

@@ -0,0 +1,7 @@
+#[derive(serde::Serialize, serde::Deserialize, Copy, Clone, Debug)]
+pub enum LogSeverity {
+    Debug,
+    Info,
+    Warn,
+    Error,
+}

central_backend/src/main.rs

@@ -19,6 +19,7 @@ async fn main() -> std::io::Result<()> {
     create_directory_if_missing(AppConfig::get().devices_config_path()).unwrap();
     create_directory_if_missing(AppConfig::get().relays_runtime_stats_storage_path()).unwrap();
     create_directory_if_missing(AppConfig::get().energy_consumption_history()).unwrap();
+    create_directory_if_missing(AppConfig::get().logs_dir()).unwrap();
 
     // Initialize PKI
     pki::initialize_root_ca().expect("Failed to initialize Root CA!");

central_backend/src/server/devices_api/device_logging_controller.rs

@@ -0,0 +1,22 @@
+use crate::logs::logs_manager;
+use crate::logs::severity::LogSeverity;
+use crate::server::custom_error::HttpResult;
+use crate::server::devices_api::jwt_parser::JWTRequest;
+use crate::server::WebEnergyActor;
+use actix_web::{web, HttpResponse};
+
+#[derive(Debug, serde::Deserialize)]
+pub struct LogRequest {
+    severity: LogSeverity,
+    message: String,
+}
+
+/// Report log message from device
+pub async fn report_log(body: web::Json<JWTRequest>, actor: WebEnergyActor) -> HttpResult {
+    let (device, request) = body.parse_jwt::<LogRequest>(actor).await?;
+
+    log::info!("Save log message from device: {request:#?}");
+    logs_manager::save_log(Some(&device.id), request.severity, request.message)?;
+
+    Ok(HttpResponse::Accepted().finish())
+}

central_backend/src/server/devices_api/jwt_parser.rs

@@ -0,0 +1,96 @@
+use crate::app_config::AppConfig;
+use crate::crypto::pki;
+use crate::devices::device::{Device, DeviceId};
+use crate::energy::energy_actor;
+use crate::server::WebEnergyActor;
+use jsonwebtoken::{Algorithm, DecodingKey, Validation};
+use openssl::x509::X509;
+use serde::de::DeserializeOwned;
+use std::collections::HashSet;
+
+#[derive(thiserror::Error, Debug)]
+pub enum JWTError {
+    #[error("Failed to decode JWT header")]
+    FailedDecodeJWT,
+    #[error("Missing KID in JWT!")]
+    MissingKidInJWT,
+    #[error("Sent a JWT for a device which does not exists!")]
+    DeviceDoesNotExists,
+    #[error("Sent a JWT for a device which is not validated!")]
+    DeviceNotValidated,
+    #[error("Sent a JWT using a revoked certificate!")]
+    RevokedCertificate,
+    #[error("Failed to validate JWT!")]
+    FailedValidateJWT,
+}
+
+#[derive(serde::Deserialize)]
+pub struct JWTRequest {
+    pub payload: String,
+}
+
+impl JWTRequest {
+    pub async fn parse_jwt<E: DeserializeOwned>(
+        &self,
+        actor: WebEnergyActor,
+    ) -> anyhow::Result<(Device, E)> {
+        // First, we need to extract device kid from query
+        let Ok(jwt_header) = jsonwebtoken::decode_header(&self.payload) else {
+            log::error!("Failed to decode JWT header!");
+            return Err(JWTError::FailedDecodeJWT.into());
+        };
+
+        let Some(kid) = jwt_header.kid else {
+            log::error!("Missing KID in JWT!");
+            return Err(JWTError::MissingKidInJWT.into());
+        };
+
+        // Fetch device information
+        let Some(device) = actor
+            .send(energy_actor::GetSingleDevice(DeviceId(kid)))
+            .await?
+        else {
+            log::error!("Sent a JWT for a device which does not exists!");
+            return Err(JWTError::DeviceDoesNotExists.into());
+        };
+
+        if !device.validated {
+            log::error!("Sent a JWT for a device which is not validated!");
+            return Err(JWTError::DeviceNotValidated.into());
+        }
+
+        // Check certificate revocation status
+        let cert_bytes = std::fs::read(AppConfig::get().device_cert_path(&device.id))?;
+        let certificate = X509::from_pem(&cert_bytes)?;
+
+        if pki::CertData::load_devices_ca()?.is_revoked(&certificate)? {
+            log::error!("Sent a JWT using a revoked certificate!");
+            return Err(JWTError::RevokedCertificate.into());
+        }
+
+        let (key, alg) = match DecodingKey::from_ec_pem(&cert_bytes) {
+            Ok(key) => (key, Algorithm::ES256),
+            Err(e) => {
+                log::warn!("Failed to decode certificate as EC certificate {e}, trying RSA...");
+                (
+                    DecodingKey::from_rsa_pem(&cert_bytes)
+                        .expect("Failed to decode RSA certificate"),
+                    Algorithm::RS256,
+                )
+            }
+        };
+        let mut validation = Validation::new(alg);
+        validation.validate_exp = false;
+        validation.required_spec_claims = HashSet::default();
+
+        let c = match jsonwebtoken::decode::<E>(&self.payload, &key, &validation) {
+            Ok(c) => c,
+            Err(e) => {
+                log::error!("Failed to validate JWT! {e}");
+                return Err(JWTError::FailedValidateJWT.into());
+            }
+        };
+
+        Ok((device, c.claims))
+    }
+}

central_backend/src/server/devices_api/mgmt_controller.rs

@@ -1,15 +1,13 @@
 use crate::app_config::AppConfig;
-use crate::crypto::pki;
 use crate::devices::device::{DeviceId, DeviceInfo};
 use crate::energy::energy_actor;
 use crate::energy::energy_actor::RelaySyncStatus;
 use crate::server::custom_error::HttpResult;
+use crate::server::devices_api::jwt_parser::JWTRequest;
 use crate::server::WebEnergyActor;
 use actix_web::{web, HttpResponse};
-use jsonwebtoken::{Algorithm, DecodingKey, Validation};
 use openssl::nid::Nid;
-use openssl::x509::{X509Req, X509};
+use openssl::x509::X509Req;
-use std::collections::HashSet;
 
 #[derive(Debug, serde::Deserialize)]
 pub struct EnrollRequest {
@@ -129,11 +127,6 @@ pub async fn get_certificate(query: web::Query<ReqWithDevID>, actor: WebEnergyAc
         .body(cert))
 }
 
-#[derive(serde::Deserialize)]
-pub struct SyncRequest {
-    payload: String,
-}
-
 #[derive(Debug, serde::Serialize, serde::Deserialize)]
 struct Claims {
     info: DeviceInfo,
@@ -145,68 +138,11 @@ struct SyncResult {
 }
 
 /// Synchronize device
-pub async fn sync_device(body: web::Json<SyncRequest>, actor: WebEnergyActor) -> HttpResult {
+pub async fn sync_device(body: web::Json<JWTRequest>, actor: WebEnergyActor) -> HttpResult {
-    // First, we need to extract device kid from query
+    let (device, claims) = body.0.parse_jwt::<Claims>(actor.clone()).await?;
-    let Ok(jwt_header) = jsonwebtoken::decode_header(&body.payload) else {
-        log::error!("Failed to decode JWT header!");
-        return Ok(HttpResponse::BadRequest().json("Failed to decode JWT header!"));
-    };
-
-    let Some(kid) = jwt_header.kid else {
-        log::error!("Missing KID in JWT!");
-        return Ok(HttpResponse::BadRequest().json("Missing KID in JWT!"));
-    };
-
-    // Fetch device information
-    let Some(device) = actor
-        .send(energy_actor::GetSingleDevice(DeviceId(kid)))
-        .await?
-    else {
-        log::error!("Sent a JWT for a device which does not exists!");
-        return Ok(HttpResponse::NotFound().json("Sent a JWT for a device which does not exists!"));
-    };
-
-    if !device.validated {
-        log::error!("Sent a JWT for a device which is not validated!");
-        return Ok(HttpResponse::PreconditionFailed()
-            .json("Sent a JWT for a device which is not validated!"));
-    }
-
-    // Check certificate revocation status
-    let cert_bytes = std::fs::read(AppConfig::get().device_cert_path(&device.id))?;
-    let certificate = X509::from_pem(&cert_bytes)?;
-
-    if pki::CertData::load_devices_ca()?.is_revoked(&certificate)? {
-        log::error!("Sent a JWT using a revoked certificate!");
-        return Ok(
-            HttpResponse::PreconditionFailed().json("Sent a JWT using a revoked certificate!")
-        );
-    }
-
-    let (key, alg) = match DecodingKey::from_ec_pem(&cert_bytes) {
-        Ok(key) => (key, Algorithm::ES256),
-        Err(e) => {
-            log::warn!("Failed to decode certificate as EC certificate {e}, trying RSA...");
-            (
-                DecodingKey::from_rsa_pem(&cert_bytes).expect("Failed to decode RSA certificate"),
-                Algorithm::RS256,
-            )
-        }
-    };
-    let mut validation = Validation::new(alg);
-    validation.validate_exp = false;
-    validation.required_spec_claims = HashSet::default();
-
-    let c = match jsonwebtoken::decode::<Claims>(&body.payload, &key, &validation) {
-        Ok(c) => c,
-        Err(e) => {
-            log::error!("Failed to validate JWT! {e}");
-            return Ok(HttpResponse::PreconditionFailed().json("Failed to validate JWT!"));
-        }
-    };
 
     let relays = actor
-        .send(energy_actor::SynchronizeDevice(device.id, c.claims.info))
+        .send(energy_actor::SynchronizeDevice(device.id, claims.info))
         .await??;
 
     Ok(HttpResponse::Ok().json(SyncResult { relays }))

central_backend/src/server/devices_api/mod.rs

@@ -1,2 +1,4 @@
+pub mod device_logging_controller;
+pub mod jwt_parser;
 pub mod mgmt_controller;
 pub mod utils_controller;

central_backend/src/server/servers.rs

@@ -3,7 +3,7 @@ use crate::constants;
 use crate::crypto::pki;
 use crate::energy::energy_actor::EnergyActorAddr;
 use crate::server::auth_middleware::AuthChecker;
-use crate::server::devices_api::{mgmt_controller, utils_controller};
+use crate::server::devices_api::{device_logging_controller, mgmt_controller, utils_controller};
 use crate::server::unsecure_server::*;
 use crate::server::web_api::*;
 use crate::server::web_app_controller;
@@ -226,6 +226,10 @@ pub async fn secure_server(energy_actor: EnergyActorAddr) -> anyhow::Result<()>
                 "/devices_api/mgmt/sync",
                 web::post().to(mgmt_controller::sync_device),
             )
+            .route(
+                "/devices_api/logging/record",
+                web::post().to(device_logging_controller::report_log),
+            )
             // Web app
             .route("/", web::get().to(web_app_controller::root_index))
             .route(

central_backend/src/utils/time_utils.rs

@@ -22,6 +22,11 @@ pub fn day_number(time: u64) -> u64 {
     time / (3600 * 24)
 }
 
+/// Get current day number
+pub fn curr_day_number() -> u64 {
+    day_number(time_secs())
+}
+
 /// Get current hour, 00 => 23 (local time)
 pub fn curr_hour() -> u32 {
     let local: DateTime<Local> = Local::now();

python_device/src/api.py

@@ -4,6 +4,7 @@ import src.constants as constants
 from cryptography.x509 import load_pem_x509_certificate
 from cryptography import utils
 import jwt
+import json
 
 
 def get_secure_origin() -> str:
@@ -75,13 +76,18 @@ def device_certificate() -> str:
     return res.text
 
 
+def jwt_sign(data: any, dev_id: str, privkey) -> str:
+    """
+    Generate a JWT for client request
+    """
+    return jwt.encode(data, privkey, algorithm="RS256", headers={"kid": dev_id})
+
+
 def sync_device(dev_id: str, privkey):
     """
     Synchronize device with backend
     """
-    encoded = jwt.encode(
+    encoded = jwt_sign({"info": device_info()}, dev_id=dev_id, privkey=privkey)
-        {"info": device_info()}, privkey, algorithm="RS256", headers={"kid": dev_id}
-    )
 
     res = requests.post(
         f"{args.secure_origin}/devices_api/mgmt/sync",
@@ -89,6 +95,19 @@ def sync_device(dev_id: str, privkey):
         verify=args.root_ca_path,
     )
 
-    print(encoded)
+    return json.loads(res.text)
-    print(res)
+
-    print(res.text)
+
+def report_log(severity: str, message: str, dev_id: str, privkey):
+    """
+    Report log message to server
+    """
+    encoded = jwt_sign(
+        {"severity": severity, "message": message}, dev_id=dev_id, privkey=privkey
+    )
+
+    requests.post(
+        f"{args.secure_origin}/devices_api/logging/record",
+        json={"payload": encoded},
+        verify=args.root_ca_path,
+    )

python_device/src/main.py

@@ -3,6 +3,10 @@ import src.api as api
 import src.pki as pki
 import src.utils as utils
 import os
+import time
+
+# TODO : turn off all relays
+# TODO : intialize GPIO
 
 print("Check storage")
 if not os.path.isdir(args.storage):
@@ -88,6 +92,13 @@ if not os.path.isfile(args.dev_crt_path):
     with open(args.dev_crt_path, "w") as f:
         f.write(cert)
 
+api.report_log("Info", "Starting program main loop...", args.dev_id, args.priv_key)
 
-print("Done. ready to operate.")
+print("Ready to operate!.")
-api.sync_device(args.dev_id, args.priv_key)
+while True:
+
+    # TODO : implement this loop more properly
+    res = api.sync_device(args.dev_id, args.priv_key)
+    print(res)
+
+    time.sleep(5)