feat: gRPC authorization for events

closes: https://todo.sr.ht/~gheartsfield/nostr-rs-relay/46

src/config.rs

@@ -25,6 +25,12 @@ pub struct Database {
     pub connection: String,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[allow(unused)]
+pub struct Grpc {
+    pub event_admission_server: Option<String>,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[allow(unused)]
 pub struct Network {
@@ -145,6 +151,7 @@ pub struct Settings {
     pub info: Info,
     pub diagnostics: Diagnostics,
     pub database: Database,
+    pub grpc: Grpc,
     pub network: Network,
     pub limits: Limits,
     pub authorization: Authorization,
@@ -220,6 +227,9 @@ impl Default for Settings {
                 max_conn: 8,
 		connection: "".to_owned(),
             },
+            grpc: Grpc {
+                event_admission_server: None,
+            },
             network: Network {
                 port: 8080,
                 ping_interval_seconds: 300,

src/db.rs

@@ -4,6 +4,7 @@ use crate::error::{Error, Result};
 use crate::event::Event;
 use crate::notice::Notice;
 use crate::server::NostrMetrics;
+use crate::nauthz;
 use governor::clock::Clock;
 use governor::{Quota, RateLimiter};
 use r2d2;
@@ -27,6 +28,8 @@ pub struct SubmittedEvent {
     pub event: Event,
     pub notice_tx: tokio::sync::mpsc::Sender<Notice>,
     pub source_ip: String,
+    pub origin: Option<String>,
+    pub user_agent: Option<String>,
 }
 
 /// Database file
@@ -101,6 +104,18 @@ pub async fn db_writer(
             lim_opt = Some(RateLimiter::direct(Quota::per_minute(quota)));
         }
     }
+    // create a client if GRPC is enabled.
+    // Check with externalized event admitter service, if one is defined.
+    let mut grpc_client = if let Some(svr) = settings.grpc.event_admission_server {
+        Some(nauthz::EventAuthzService::connect(&svr).await)
+    } else {
+        None
+    };
+
+    //let gprc_client = settings.grpc.event_admission_server.map(|s| {
+//        event_admitter_connect(&s);
+//    });
+
     loop {
         if shutdown.try_recv().is_ok() {
             info!("shutting down database writer");
@@ -165,9 +180,17 @@ pub async fn db_writer(
             metadata_tx.send(event.clone()).ok();
         }
 
+        // get a validation result for use in verification and GPRC
+        let validation = if nip05_active {
+            Some(repo.get_latest_user_verification(&event.pubkey).await)
+        } else {
+            None
+        };
+
+
         // check for  NIP-05 verification
-        if nip05_enabled {
-            match repo.get_latest_user_verification(&event.pubkey).await {
+        if nip05_enabled && validation.is_some() {
+            match validation.as_ref().unwrap() {
                 Ok(uv) => {
                     if uv.is_valid(&settings.verified_users) {
                         info!(
@@ -175,6 +198,7 @@ pub async fn db_writer(
                             uv.name.to_string(),
                             event.get_author_prefix()
                         );
+
                     } else {
                         info!(
                             "rejecting event, author ({:?} / {:?}) verification invalid (expired/wrong domain)",
@@ -209,6 +233,35 @@ pub async fn db_writer(
                 }
             }
         }
+
+        // nip05 address
+        let nip05_address : Option<crate::nip05::Nip05Name> = validation.and_then(|x| x.ok().map(|y| y.name));
+
+        // GRPC check
+        if let Some(ref mut c) = grpc_client {
+            trace!("checking if grpc permits");
+            let grpc_start = Instant::now();
+            let decision_res = c.admit_event(&event, &subm_event.source_ip, subm_event.origin, subm_event.user_agent, nip05_address).await;
+            match decision_res {
+                Ok(decision) => {
+                    if !decision.permitted() {
+                        // GPRC returned a decision to reject this event
+                        info!("GRPC rejected event: {:?} (kind: {}) from: {:?} in: {:?} (IP: {:?})",
+                              event.get_event_id_prefix(),
+                              event.kind,
+                              event.get_author_prefix(),
+                              grpc_start.elapsed(),
+                              subm_event.source_ip);
+                        notice_tx.try_send(Notice::blocked(event.id, &decision.message().unwrap_or_else(|| "".to_string()))).ok();
+                        continue;
+                    }
+                },
+                Err(e) => {
+                    warn!("GRPC server error: {:?}", e);
+                }
+            }
+        }
+
         // TODO: cache recent list of authors to remove a DB call.
         let start = Instant::now();
         if event.is_ephemeral() {

src/error.rs

@@ -64,6 +64,10 @@ pub enum Error {
     DelegationParseError,
     #[error("Channel closed error")]
     ChannelClosed,
+    #[error("Authz error")]
+    AuthzError,
+    #[error("Tonic GRPC error")]
+    TonicError(tonic::Status),
     #[error("Unknown/Undocumented")]
     UnknownError,
 }
@@ -132,3 +136,10 @@ impl From<config::ConfigError> for Error {
         Error::ConfigError(r)
     }
 }
+
+impl From<tonic::Status> for Error {
+    /// Wrap Config error
+    fn from(r: tonic::Status) -> Self {
+        Error::TonicError(r)
+    }
+}

src/lib.rs

@@ -9,6 +9,7 @@ pub mod event;
 pub mod hexrange;
 pub mod info;
 pub mod nip05;
+pub mod nauthz;
 pub mod notice;
 pub mod repo;
 pub mod subscription;

src/nauthz.rs

@@ -0,0 +1,110 @@
+use crate::error::{Error, Result};
+use crate::{event::Event, nip05::Nip05Name};
+use nauthz_grpc::authorization_client::AuthorizationClient;
+use nauthz_grpc::event::TagEntry;
+use nauthz_grpc::{Decision, Event as GrpcEvent, EventReply, EventRequest};
+use tracing::{info, warn};
+
+pub mod nauthz_grpc {
+    tonic::include_proto!("nauthz");
+}
+
+// A decision for the DB to act upon
+pub trait AuthzDecision: Send + Sync {
+    fn permitted(&self) -> bool;
+    fn message(&self) -> Option<String>;
+}
+
+impl AuthzDecision for EventReply {
+    fn permitted(&self) -> bool {
+        self.decision == Decision::Permit as i32
+    }
+    fn message(&self) -> Option<String> {
+        self.message.clone()
+    }
+}
+
+// A connection to an event admission GRPC server
+pub struct EventAuthzService {
+    server_addr: String,
+    conn: Option<AuthorizationClient<tonic::transport::Channel>>,
+}
+
+// conversion of Nip05Names into GRPC type
+impl std::convert::From<Nip05Name> for nauthz_grpc::event_request::Nip05Name {
+    fn from(value: Nip05Name) -> Self {
+        nauthz_grpc::event_request::Nip05Name {
+            local: value.local.clone(),
+            domain: value.domain.clone(),
+        }
+    }
+}
+
+// conversion of event tags into gprc struct
+fn tags_to_protobuf(tags: &Vec<Vec<String>>) -> Vec<TagEntry> {
+    tags.iter()
+        .map(|x| TagEntry { values: x.clone() })
+        .collect()
+}
+
+impl EventAuthzService {
+    pub async fn connect(server_addr: &str) -> EventAuthzService {
+        let mut eas = EventAuthzService {
+            server_addr: server_addr.to_string(),
+            conn: None,
+        };
+        eas.ready_connection().await;
+        eas
+    }
+
+    pub async fn ready_connection(self: &mut Self) {
+        if self.conn.is_none() {
+            let client = AuthorizationClient::connect(self.server_addr.to_string()).await;
+            if let Err(ref msg) = client {
+                warn!("could not connect to nostr authz GRPC server: {:?}", msg);
+            } else {
+                info!("connected to nostr authorization GRPC server");
+            }
+            self.conn = client.ok();
+        }
+    }
+
+    pub async fn admit_event(
+        self: &mut Self,
+        event: &Event,
+        ip: &str,
+        origin: Option<String>,
+        user_agent: Option<String>,
+        nip05: Option<Nip05Name>,
+    ) -> Result<Box<dyn AuthzDecision>> {
+        self.ready_connection().await;
+        let id_blob = hex::decode(&event.id)?;
+        let pubkey_blob = hex::decode(&event.pubkey)?;
+        let sig_blob = hex::decode(&event.sig)?;
+        if let Some(ref mut c) = self.conn {
+            let gevent = GrpcEvent {
+                id: id_blob,
+                pubkey: pubkey_blob,
+                sig: sig_blob,
+                created_at: event.created_at,
+                kind: event.kind,
+                content: event.content.clone(),
+                tags: tags_to_protobuf(&event.tags),
+            };
+            let svr_res = c
+                .event_admit(EventRequest {
+                    event: Some(gevent),
+                    ip_addr: Some(ip.to_string()),
+                    origin,
+                    user_agent,
+                    auth_pubkey: None,
+                    nip05: nip05.map(|x| nauthz_grpc::event_request::Nip05Name::from(x)),
+                })
+                .await?;
+            let reply = svr_res.into_inner();
+            return Ok(Box::new(reply));
+        } else {
+            return Err(Error::AuthzError);
+        }
+    }
+}

src/nip05.rs

@@ -42,8 +42,8 @@ pub struct Verifier {
 /// A NIP-05 identifier is a local part and domain.
 #[derive(PartialEq, Eq, Debug, Clone)]
 pub struct Nip05Name {
-    local: String,
-    domain: String,
+    pub local: String,
+    pub domain: String,
 }
 
 impl Nip05Name {

src/server.rs

@@ -601,11 +601,13 @@ async fn nostr_server(
     // and how many it received from queries.
     let mut client_published_event_count: usize = 0;
     let mut client_received_event_count: usize = 0;
+
+    let unspec = "<unspecified>".to_string();
     info!("new client connection (cid: {}, ip: {:?})", cid, conn.ip());
-    let origin = client_info.origin.unwrap_or_else(|| "<unspecified>".into());
+    let origin = client_info.origin.as_ref().unwrap_or_else(|| &unspec);
     let user_agent = client_info
-        .user_agent
-        .unwrap_or_else(|| "<unspecified>".into());
+        .user_agent.as_ref()
+        .unwrap_or_else(|| &unspec);
     info!(
         "cid: {}, origin: {:?}, user-agent: {:?}",
         cid, origin, user_agent
@@ -736,7 +738,7 @@ async fn nostr_server(
                                 // check if the event is too far in the future.
                                 if e.is_valid_timestamp(settings.options.reject_future_seconds) {
                                     // Write this to the database.
-                                    let submit_event = SubmittedEvent { event: e.clone(), notice_tx: notice_tx.clone(), source_ip: conn.ip().to_string()};
+                                    let submit_event = SubmittedEvent { event: e.clone(), notice_tx: notice_tx.clone(), source_ip: conn.ip().to_string(), origin: client_info.origin.clone(), user_agent: client_info.user_agent.clone()};
                                     event_tx.send(submit_event).await.ok();
                                     client_published_event_count += 1;
                                 } else {