-
Notifications
You must be signed in to change notification settings - Fork 25
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
mobile verifier support for invalidated threshold reports (#754)
* ingestor support for hotspot thresholds add cbsd_id support refactor for proto msg renames * bump proto * bump proto * mobile verifier support for hotspot thresholds recut branch & add support for cbsd_id in hotspot thresholds handle grandfathering * bump proto * tweaks * review tweaks/fixes * Update mobile_verifier/src/reward_shares.rs Co-authored-by: Matthew Plant <[email protected]> * fix it * ingestor and filestore support for invalidated threshold reports * handle invalidated threshold reports * tweaks * rebase artifacts * tweaks --------- Co-authored-by: Matthew Plant <[email protected]> Co-authored-by: Matthew Plant <[email protected]>
- Loading branch information
3 people
authored
Mar 7, 2024
1 parent
3694c31
commit 6d46823
Showing
6 changed files
with
204 additions
and
5 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,160 @@ | ||
| use chrono::Utc; | ||
| use file_store::{ | ||
| file_info_poller::FileInfoStream, | ||
| file_sink::FileSinkClient, | ||
| mobile_radio_invalidated_threshold::{ | ||
| InvalidatedRadioThresholdIngestReport, InvalidatedRadioThresholdReportReq, | ||
| VerifiedInvalidatedRadioThresholdIngestReport, | ||
| }, | ||
| }; | ||
| use futures::{StreamExt, TryStreamExt}; | ||
| use futures_util::TryFutureExt; | ||
| use helium_crypto::PublicKeyBinary; | ||
| use helium_proto::services::{ | ||
| mobile_config::NetworkKeyRole, | ||
| poc_mobile::{ | ||
| InvalidatedRadioThresholdReportVerificationStatus, | ||
| VerifiedInvalidatedRadioThresholdIngestReportV1, | ||
| }, | ||
| }; | ||
| use mobile_config::client::authorization_client::AuthorizationVerifier; | ||
| use sqlx::{PgPool, Postgres, Transaction}; | ||
| use task_manager::ManagedTask; | ||
| use tokio::sync::mpsc::Receiver; | ||
|
|
||
| pub struct InvalidatedRadioThresholdIngestor<AV> { | ||
| pool: PgPool, | ||
| reports_receiver: Receiver<FileInfoStream<InvalidatedRadioThresholdIngestReport>>, | ||
| verified_report_sink: FileSinkClient, | ||
| authorization_verifier: AV, | ||
| } | ||
|
|
||
| impl<AV> ManagedTask for InvalidatedRadioThresholdIngestor<AV> | ||
| where | ||
| AV: AuthorizationVerifier + Send + Sync + 'static, | ||
| { | ||
| fn start_task( | ||
| self: Box<Self>, | ||
| shutdown: triggered::Listener, | ||
| ) -> futures_util::future::LocalBoxFuture<'static, anyhow::Result<()>> { | ||
| let handle = tokio::spawn(self.run(shutdown)); | ||
| Box::pin( | ||
| handle | ||
| .map_err(anyhow::Error::from) | ||
| .and_then(|result| async move { result.map_err(anyhow::Error::from) }), | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| impl<AV> InvalidatedRadioThresholdIngestor<AV> | ||
| where | ||
| AV: AuthorizationVerifier, | ||
| { | ||
| pub fn new( | ||
| pool: sqlx::Pool<Postgres>, | ||
| reports_receiver: Receiver<FileInfoStream<InvalidatedRadioThresholdIngestReport>>, | ||
| verified_report_sink: FileSinkClient, | ||
| authorization_verifier: AV, | ||
| ) -> Self { | ||
| Self { | ||
| pool, | ||
| reports_receiver, | ||
| verified_report_sink, | ||
| authorization_verifier, | ||
| } | ||
| } | ||
|
|
||
| async fn run(mut self, shutdown: triggered::Listener) -> anyhow::Result<()> { | ||
| tracing::info!("starting invalidated radio threshold ingestor"); | ||
| loop { | ||
| tokio::select! { | ||
| biased; | ||
| _ = shutdown.clone() => break, | ||
| Some(file) = self.reports_receiver.recv() => { | ||
| self.process_file(file).await?; | ||
| } | ||
| } | ||
| } | ||
| tracing::info!("stopping invalidated radio threshold ingestor"); | ||
| Ok(()) | ||
| } | ||
|
|
||
| async fn process_file( | ||
| &self, | ||
| file_info_stream: FileInfoStream<InvalidatedRadioThresholdIngestReport>, | ||
| ) -> anyhow::Result<()> { | ||
| let mut transaction = self.pool.begin().await?; | ||
| file_info_stream | ||
| .into_stream(&mut transaction) | ||
| .await? | ||
| .map(anyhow::Ok) | ||
| .try_fold(transaction, |mut transaction, ingest_report| async move { | ||
| // verify the report | ||
| let verified_report_status = self.verify_report(&ingest_report.report).await; | ||
|
|
||
| // if the report is valid then delete the thresholds from the DB | ||
| if verified_report_status == InvalidatedRadioThresholdReportVerificationStatus::InvalidatedThresholdReportStatusValid { | ||
| delete(&ingest_report, &mut transaction).await?; | ||
| } | ||
|
|
||
| // write out paper trail of verified report, valid or invalid | ||
| let verified_report_proto: VerifiedInvalidatedRadioThresholdIngestReportV1 = | ||
| VerifiedInvalidatedRadioThresholdIngestReport { | ||
| report: ingest_report, | ||
| status: verified_report_status, | ||
| timestamp: Utc::now(), | ||
| } | ||
| .into(); | ||
| self.verified_report_sink | ||
| .write( | ||
| verified_report_proto, | ||
| &[("report_status", verified_report_status.as_str_name())], | ||
| ) | ||
| .await?; | ||
| Ok(transaction) | ||
| }) | ||
| .await? | ||
| .commit() | ||
| .await?; | ||
| self.verified_report_sink.commit().await?; | ||
| Ok(()) | ||
| } | ||
|
|
||
| async fn verify_report( | ||
| &self, | ||
| report: &InvalidatedRadioThresholdReportReq, | ||
| ) -> InvalidatedRadioThresholdReportVerificationStatus { | ||
| if !self.verify_known_carrier_key(&report.carrier_pub_key).await { | ||
| return InvalidatedRadioThresholdReportVerificationStatus::InvalidatedThresholdReportStatusInvalidCarrierKey; | ||
| }; | ||
| InvalidatedRadioThresholdReportVerificationStatus::InvalidatedThresholdReportStatusValid | ||
| } | ||
|
|
||
| async fn verify_known_carrier_key(&self, public_key: &PublicKeyBinary) -> bool { | ||
| match self | ||
| .authorization_verifier | ||
| .verify_authorized_key(public_key, NetworkKeyRole::MobileCarrier) | ||
| .await | ||
| { | ||
| Ok(res) => res, | ||
| Err(_err) => false, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| pub async fn delete( | ||
| ingest_report: &InvalidatedRadioThresholdIngestReport, | ||
| db: &mut Transaction<'_, Postgres>, | ||
| ) -> Result<(), sqlx::Error> { | ||
| sqlx::query( | ||
| r#" | ||
| DELETE FROM radio_threshold | ||
| WHERE hotspot_pubkey = $1 AND (cbsd_id is null or cbsd_id = $2) | ||
| "#, | ||
| ) | ||
| .bind(ingest_report.report.hotspot_pubkey.to_string()) | ||
| .bind(ingest_report.report.cbsd_id.clone()) | ||
| .execute(&mut *db) | ||
| .await?; | ||
| Ok(()) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters