Merge pull request #546 from helium/jg/mobile-config-subscriber-rewards

Mobile config service supports subscriber rewards

file_store/src/traits/msg_verify.rs

@@ -39,6 +39,10 @@ impl_msg_verify!(iot_config::OrgCreateRoamerReqV1, signature);
 impl_msg_verify!(iot_config::OrgUpdateReqV1, signature);
 impl_msg_verify!(iot_config::OrgDisableReqV1, signature);
 impl_msg_verify!(iot_config::OrgEnableReqV1, signature);
+impl_msg_verify!(iot_config::OrgDisableResV1, signature);
+impl_msg_verify!(iot_config::OrgEnableResV1, signature);
+impl_msg_verify!(iot_config::OrgResV1, signature);
+impl_msg_verify!(iot_config::OrgListResV1, signature);
 impl_msg_verify!(iot_config::RouteStreamReqV1, signature);
 impl_msg_verify!(iot_config::RouteListReqV1, signature);
 impl_msg_verify!(iot_config::RouteGetReqV1, signature);
@@ -65,8 +69,12 @@ impl_msg_verify!(iot_config::GatewayInfoStreamResV1, signature);
 impl_msg_verify!(iot_config::RegionParamsResV1, signature);
 impl_msg_verify!(mobile_config::AdminAddKeyReqV1, signature);
 impl_msg_verify!(mobile_config::AdminRemoveKeyReqV1, signature);
-impl_msg_verify!(mobile_config::RouterGetReqV1, signature);
-impl_msg_verify!(mobile_config::RouterListReqV1, signature);
+impl_msg_verify!(mobile_config::AuthorizationVerifyReqV1, signature);
+impl_msg_verify!(mobile_config::AuthorizationVerifyResV1, signature);
+impl_msg_verify!(mobile_config::AuthorizationListReqV1, signature);
+impl_msg_verify!(mobile_config::AuthorizationListResV1, signature);
+impl_msg_verify!(mobile_config::EntityVerifyReqV1, signature);
+impl_msg_verify!(mobile_config::EntityVerifyResV1, signature);
 impl_msg_verify!(mobile_config::GatewayInfoReqV1, signature);
 impl_msg_verify!(mobile_config::GatewayInfoStreamReqV1, signature);
 impl_msg_verify!(mobile_config::GatewayInfoResV1, signature);

iot_config/src/client/mod.rs

@@ -8,7 +8,10 @@ use helium_proto::{
 };
 use std::{sync::Arc, time::Duration};
 
+pub mod org_client;
 mod settings;
+
+pub use org_client::OrgClient;
 pub use settings::Settings;
 
 #[derive(thiserror::Error, Debug)]

iot_config/src/client/org_client.rs

@@ -0,0 +1,77 @@
+use super::{
+    iot_config, Arc, Channel, ClientError, Duration, Endpoint, Keypair, Message, MsgVerify,
+    PublicKey, Settings, Sign,
+};
+use chrono::Utc;
+use file_store::traits::TimestampEncode;
+use helium_proto::services::iot_config::{
+    OrgDisableReqV1, OrgEnableReqV1, OrgGetReqV1, OrgListReqV1, OrgResV1, OrgV1,
+};
+
+#[derive(Clone)]
+pub struct OrgClient {
+    client: iot_config::config_org_client::OrgClient<Channel>,
+    signing_key: Arc<Keypair>,
+    config_pubkey: PublicKey,
+}
+
+impl OrgClient {
+    pub fn from_settings(settings: &Settings) -> Result<Self, Box<helium_crypto::Error>> {
+        let channel = Endpoint::from(settings.url.clone())
+            .connect_timeout(Duration::from_secs(settings.connect_timeout))
+            .timeout(Duration::from_secs(settings.rpc_timeout))
+            .connect_lazy();
+        Ok(Self {
+            client: iot_config::config_org_client::OrgClient::new(channel),
+            signing_key: settings.signing_keypair()?,
+            config_pubkey: settings.config_pubkey()?,
+        })
+    }
+
+    pub async fn get(&mut self, oui: u64) -> Result<OrgResV1, ClientError> {
+        tracing::debug!(%oui, "retrieving org");
+
+        let req = OrgGetReqV1 { oui };
+        let res = self.client.get(req).await?.into_inner();
+        res.verify(&self.config_pubkey)?;
+        Ok(res)
+    }
+
+    pub async fn list(&mut self) -> Result<Vec<OrgV1>, ClientError> {
+        tracing::debug!("retrieving org list");
+
+        let res = self.client.list(OrgListReqV1 {}).await?.into_inner();
+        res.verify(&self.config_pubkey)?;
+        Ok(res.orgs)
+    }
+
+    pub async fn enable(&mut self, oui: u64) -> Result<(), ClientError> {
+        tracing::info!(%oui, "enabling org");
+
+        let mut req = OrgEnableReqV1 {
+            oui,
+            timestamp: Utc::now().encode_timestamp(),
+            signer: self.signing_key.public_key().into(),
+            signature: vec![],
+        };
+        req.signature = self.signing_key.sign(&req.encode_to_vec())?;
+        let res = self.client.enable(req).await?.into_inner();
+        res.verify(&self.config_pubkey)?;
+        Ok(())
+    }
+
+    pub async fn disable(&mut self, oui: u64) -> Result<(), ClientError> {
+        tracing::info!(%oui, "disabling org");
+
+        let mut req = OrgDisableReqV1 {
+            oui,
+            timestamp: Utc::now().encode_timestamp(),
+            signer: self.signing_key.public_key().into(),
+            signature: vec![],
+        };
+        req.signature = self.signing_key.sign(&req.encode_to_vec())?;
+        let res = self.client.disable(req).await?.into_inner();
+        res.verify(&self.config_pubkey)?;
+        Ok(())
+    }
+}

iot_config/src/org.rs

@@ -142,7 +142,9 @@ pub async fn create_org(
     }
     .map_err(|err| OrgStoreError::SaveConstraints(format!("{devaddr_ranges:?}: {err:?}")))?;
 
-    let org = get(oui as u64, &mut txn).await?;
+    let org = get(oui as u64, &mut txn)
+        .await?
+        .ok_or_else(|| OrgStoreError::SaveOrg(format!("{oui}")))?;
 
     txn.commit().await?;
 
@@ -157,9 +159,11 @@ pub async fn update_org(
 ) -> Result<Org, OrgStoreError> {
     let mut txn = db.begin().await?;
 
+    let current_org = get(oui, &mut txn)
+        .await?
+        .ok_or_else(|| OrgStoreError::NotFound(format!("{oui}")))?;
     let net_id = get_org_netid(oui, &mut txn).await?;
     let is_helium_org = is_helium_netid(&net_id);
-    let current_org = get(oui, &mut txn).await?;
 
     for update in updates {
         match update.update {
@@ -203,7 +207,9 @@ pub async fn update_org(
         };
     }
 
-    let updated_org = get(oui, &mut txn).await?;
+    let updated_org = get(oui, &mut txn)
+        .await?
+        .ok_or_else(|| OrgStoreError::SaveOrg(format!("{oui}")))?;
 
     txn.commit().await?;
 
@@ -433,13 +439,13 @@ pub async fn list(db: impl sqlx::PgExecutor<'_>) -> Result<Vec<Org>, sqlx::Error
         .await)
 }
 
-pub async fn get(oui: u64, db: impl sqlx::PgExecutor<'_>) -> Result<Org, sqlx::Error> {
+pub async fn get(oui: u64, db: impl sqlx::PgExecutor<'_>) -> Result<Option<Org>, sqlx::Error> {
     let mut query: sqlx::QueryBuilder<sqlx::Postgres> = sqlx::QueryBuilder::new(GET_ORG_SQL);
     query.push(" where org.oui = $1 ");
     query
         .build_query_as::<Org>()
         .bind(oui as i64)
-        .fetch_one(db)
+        .fetch_optional(db)
         .await
 }
 
@@ -525,6 +531,8 @@ pub async fn toggle_locked(oui: u64, db: impl sqlx::PgExecutor<'_>) -> Result<()
 pub enum OrgStoreError {
     #[error("error retrieving saved org row: {0}")]
     FetchOrg(#[from] sqlx::Error),
+    #[error("org not found: {0}")]
+    NotFound(String),
     #[error("error saving org: {0}")]
     SaveOrg(String),
     #[error("error saving delegate keys: {0}")]
@@ -543,7 +551,9 @@ pub async fn get_org_pubkeys(
     oui: u64,
     db: impl sqlx::PgExecutor<'_>,
 ) -> Result<Vec<PublicKey>, OrgStoreError> {
-    let org = get(oui, db).await?;
+    let org = get(oui, db)
+        .await?
+        .ok_or_else(|| OrgStoreError::NotFound(format!("{oui}")))?;
 
     let mut pubkeys: Vec<PublicKey> = vec![
         PublicKey::try_from(org.owner)?,

iot_config/src/org_service.rs

@@ -91,6 +91,8 @@ impl OrgService {
 
         let org_owner = org::get(request.oui, &self.pool)
             .await
+            .transpose()
+            .ok_or_else(|| Status::not_found(format!("oui: {}", request.oui)))?
             .map(|org| org.owner)
             .map_err(|_| Status::internal("auth verification error"))?;
         if org_owner == signer.clone().into() && request.verify(signer).is_ok() {
@@ -138,10 +140,13 @@ impl iot_config::Org for OrgService {
         let request = request.into_inner();
         telemetry::count_request("org", "get");
 
-        let org = org::get(request.oui, &self.pool).await.map_err(|err| {
-            tracing::error!(oui = request.oui, reason = ?err, "get org request failed");
-            Status::internal("org get failed")
-        })?;
+        let org = org::get(request.oui, &self.pool)
+            .await
+            .map_err(|err| {
+                tracing::error!(oui = request.oui, reason = ?err, "get org request failed");
+                Status::internal("org get failed")
+            })?
+            .ok_or_else(|| Status::not_found(format!("oui: {}", request.oui)))?;
         let net_id = org::get_org_netid(org.oui, &self.pool)
             .await
             .map_err(|err| {

iot_packet_verifier/Cargo.toml

@@ -18,6 +18,7 @@ futures-util = {workspace = true}
 file-store = {path = "../file_store"}
 helium-proto = {workspace = true}
 helium-crypto = {workspace = true, features = ["sqlx-postgres", "multisig", "solana"]}
+iot-config = {path = "../iot_config"}
 metrics = {workspace = true}
 poc-metrics = {path = "../metrics"}
 prost = {workspace = true}
@@ -32,7 +33,3 @@ tracing-subscriber = {workspace = true}
 triggered = {workspace = true}
 http = {workspace = true}
 http-serde = {workspace = true}
-
-
-
-