tag2upload_service_manager/

webhook.rs

//! webhook handling (forge-agnostic)
//!
//! Processing happens as follows:
//!
//!  * HHTP POST data
//!    1. processed by Rocket route handler
//!    2. `FromData for RoutePayloadParameter<F>`
//!       * Checks the vhost.
//!       * Extracts the client IP address and checks it
//!        against the union of the forge ACLs.
//!       * Deserialises the JSON into serde_json::Value.
//!    2. Per-forge webhook implementation route calls
//!       `RoutePayloadParameter<F>::webhook_impl`.
//!    3. Global implementation `webhook_impl`, here:
//!       * Goes via `webhook_impl_logged` and
//!         `RawPayload.decode_analyse_validate_record`
//!         to structure the logging and error handling.
//!       * Calls the forge-specific prefilter.
//!       * Further deserialises the payload into
//!         the forge-specific data `We3bhookForge::Payload`.
//!       * Calls `ForgeKind::analyse_payload` to get a
//!         `webhook::AnalysedPayloadData`.
//!       * Checks the host name and IP address against the specific ACL.
//!       * Parses the tag metadata.
//!       * Makes a `JobRow` and Inserts it into the database.
use crate::prelude::*;

use rocket::data::{Data, FromData, Outcome};
use rocket::request::Request;
use rocket::serde::json::Json;

/// Used in `#[rocket::post]`
pub struct RoutePayloadParameter<F> {
    payload: Result<RawPayload, WebError>,
    forge: PhantomData<F>,
}

/// Payload, JSON deserialised but not yet decoded by forge impl
struct RawPayload {
    client: ActualClient,
    #[allow(dead_code)]
    unified_acl_checked: IsAllowedClient,
    #[allow(dead_code)]
    vhost_checked: ui_vhost::IsWebhook,
    raw: serde_json::Value,
}

#[derive(Deftly, Debug)]
pub struct AnalysedPayload<F: ForgeReceiver> {
    pub repo_git_url: String,
    pub tag_name: String,
    pub tag_objectid: GitObjectId,
    pub tag_message: String,
    pub forge_data: F::DbData,
}

struct RequestBeingProcessed<'a, F: ForgeReceiver> {
    globals: &'a Globals,
    forge: &'a F,
    payload: AnalysedPayload<F>,
    raw: &'a RawPayload,
}

/// An error E that has been logged, iff it needed to be
struct Logged<E>(E);

impl Display for Logged<WebError> {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "{}, {}", self.0.http_status(), self.0)
    }
}

impl<'r> RawPayload {
    /// First non-Rocket call site for a webhook POST request
    ///
    /// Here we try to handle the POST data, but only if we decide
    /// that's appropriate.  We might just return `Err`.
    async fn from_data(
        req: &'r Request<'_>,
        data: Data<'r>,
    ) -> Result<Self, WE> {
        use Outcome as O;
        let gl = globals();

        let vhost_checked = ui_vhost::IsWebhook::from_request(req).await
            .map_err(|e| WE::PageNotFoundHere(e.into()))?;

        let client = req.client_ip()
            .ok_or_else(|| internal!("missing client IP addr"))?;
        let client = ActualClient::new(client);

        let unified_acl_checked: IsAllowedClient = client.allowed_by(
            &gl.computed_config.unified_webhook_acl
        ).await?;

        // Now we know the caller is authorised, somehow,
        // and we can reasonably safely process the POST data.

        let raw = match Json::from_data(req, data).await {
            O::Success(Json(raw)) => Ok(raw),
            O::Error((s, e)) => Err(WebError::MisconfiguredWebhook(
                anyhow!("body parsing failed ({s}): {e}")
            )),
            x @ O::Forward(_) => Err(
                internal!("forwarded?! {x:?}").into()
            ),
        }?;

        Ok(RawPayload { vhost_checked, client, unified_acl_checked, raw })
    }
}

#[async_trait]
impl<'r, F: ForgeKind> FromData<'r> for RoutePayloadParameter<F> {
    /// Errors go into `RawWebhookPayload.payload`
    ///
    /// This is because Rocket throws the error away when generating
    /// a response!
    type Error = Void;
    
    async fn from_data(
        req: &'r Request<'_>,
        data: Data<'r>,
    ) -> Outcome<'r, Self, Self::Error> {
        use Outcome as O;

        let payload = RawPayload::from_data(req, data).await;

        match &payload {
            Ok(y) => trace!(client=?y.client, raw=?y.raw, "webhook"),
            Err(err) => trace!(client=?req.client_ip(), %err, "webhook"),
        };

        let forge = PhantomData;

        O::Success(RoutePayloadParameter { payload, forge })
    }
}

impl<F: ForgeReceiver> RoutePayloadParameter<F> {
    /// Implementation of the webhook, main entrypoint
    ///
    /// All POST requests to the right path are handled here,
    /// so this function is responsible for most error handling.
    ///
    /// Errors that are checked for *before* processing POST data
    /// (including client IP addresses that aren't on any of our ACLs)
    /// show up as `self.payload` being `Err`.
    pub async fn webhook_impl(self) -> Result<String, WebError> {
        self.webhook_impl_logged().await
            .map_err(|Logged(e)| e)
    }

    /// Implementation of `webhook_impl`, with assurance of logging
    async fn webhook_impl_logged(self) -> Result<String, Logged<WebError>> {
        let raw = self.payload
            .map_err(|e| {
                let e = Logged(e);
                debug!("rejected early: {e}");
                e
            })?;

        let mut log_info = format!("from {}", &raw.client);
        let forge = F::default();

        forge.prefilter_payload(&raw.raw)
            .map_err(|e| {
                let e = Logged(e);
                debug!("filtered: {log_info}: {e}");
                e
            })?;

        raw.decode_analyse_validate_record(&forge, &mut log_info).await
            .map_err(|e| {
                let e = Logged(e);
                write_string!(log_info, ": {e}");
                match &e.0 {
                    WE::Throttled { .. } |
                    WE::DisallowedClient { .. } |
                    WE::PageNotFoundHere { .. } |
                    // reported as Error when we generated it
                    WE::InternalError { .. } => 
                        info!("rejected: {log_info}"),
                    WE::MisconfiguredWebhook { .. } |
                    WE::NotForUs { .. } => {
 // info_debug_with_raw! from decode_analyse_validate_record is not suitable
 // here, because here we have no parameters so the message text must be
 // different.  Anyway, here the repetition is much less.
                        info!("ignored: {log_info}");
                        debug!("ignored: {log_info}; raw={:?}", &raw.raw);
                    }
                }
                e
            })
    }
}

impl RawPayload {
    /// Core of webhook processing, after pre-checks have been done
    ///
    /// The caller will log errors thrown from here.
    async fn decode_analyse_validate_record<F: ForgeReceiver>(
        &self,
        forge: &F,
        log_info: &mut String,
    ) -> Result<String, WebError> {
        let payload = Deserialize::deserialize(&self.raw)
            .context("JSON payload does not conform to expected schema")
            .map_err(WE::MisconfiguredWebhook)?;

        let payload = forge.analyse_payload(payload)?;

        write_string!(log_info,
                      ", url={} tag={} objectid={}",
                      payload.repo_git_url,
                      payload.tag_name,
                      payload.tag_objectid);

        let req = RequestBeingProcessed {
            globals: &globals(),
            payload,
            forge,
            raw: self,
        };

        let forge_host = req.check_permission().await?;

        // precheck, so we don't do a bunch of work if we are throttled
        db_transaction(TN::Readonly, |dbt| check_not_throttled(dbt))??;

        let forge_namever = forge.namever().to_owned().into();

        req.check_tag_name()?;

        let AnalysedPayload {
            repo_git_url,
            tag_name,
            tag_objectid,
            forge_data,
            tag_message,
        } = req.payload;

        let tag_meta = t2umeta::Parsed::from_tag_message(&tag_message)?;

        *log_info = format!("source={}, {log_info}", tag_meta.source);

        let validated_data = JobData {
            repo_git_url,
            tag_objectid,
            tag_name,
            forge_host,
            forge_namever,
            forge_data: ForgeData::from_raw_string(forge_data.to_string()),
            tag_meta,
        };

        let now = req.globals.now();

        let initial_state = JobState {
            last_update: now,
            status: JobStatus::Noticed,
            info: format!("job received, tag not yet fetched"),
            processing: None.into(),
            duplicate_of: None,
            retry_salient_count: 0,
        };

        let job_row = JobRow {
            jid: JobId::none(),
            data: validated_data,
            received: now,
            retry_earliest: TreatZeroAsNone::none(),
            tag_data: None.into(),
            s: initial_state,
        };

        let jid = db_transaction(TN::Update { 
            this_jid: None,
            tag_objectid: &job_row.data.tag_objectid,
        }, |dbt| {
            match check_not_throttled(dbt)? {
                Ok(()) => {}
                Err(throttled) => return Ok(Err(throttled)),
            }

            let jid = dbt.bsql_insert(&bsql!(
                "INSERT INTO jobs " .insert_row(&job_row) ""
            ))?;

            Ok(Ok(jid))
        })??;

        let msg = format!("job received, jid={jid}");

        macro_rules! info_debug_with_raw { {
            [ $($sval:tt)* ], $raw:expr, $fmt:literal $($rest:tt)*
        } => {
            info !($($sval)*,            $fmt $($rest)*);
            debug!($($sval)*, raw=?$raw, $fmt $($rest)*);
        } }

        info_debug_with_raw!(
            [ jid=%jid, now=?job_row.s.status, info=?job_row.s.info,
              event=%format_args!(" {log_info}") ],
                self.raw,
            "[{}] received", job_row.data.forge_host
        );

        Ok(msg)
    }
}

impl<'a, F: ForgeReceiver> RequestBeingProcessed<'a, F> {
    async fn check_permission(&self) -> Result<Hostname, WE> {
        let forge_host = (|| {
            let rhs = if let Some(fake) = &self.globals.config
                .testing.fake_https_dir
            {
                let strip = format!("file://{fake}/");
                self.payload.repo_git_url
                    .strip_prefix(&strip)
                    .ok_or_else(|| anyhow!(
                        "failed to strip expected faked {strip:?} from {:?}",
                        self.payload.repo_git_url
                    ))?
            } else {
                self.payload.repo_git_url
                    .strip_prefix("https://")
                    .ok_or_else(|| anyhow!("scheme not https"))?
            };
            let (host, rhs) = rhs.split_once('/')
                .ok_or_else(|| anyhow!("missing / after host"))?;

            rhs.chars().all(|c| c.is_ascii_graphic()).then_some(())
                .ok_or_else(|| anyhow!("nonprintable characters in url"))?;

            let host: Hostname = host.parse()?;

            Ok::<_, AE>(host)
        })()
            .context("bad project repository URL")
            .map_err(WE::MisconfiguredWebhook)?;

        let correct_host_forges = self.globals.config.t2u.forges.iter()
            .filter(|cf| cf.host == forge_host);

        let check_kind = |cf: &config::Forge| {
            (cf.kind == self.forge.kind_name()).then(|| ())
                .ok_or_else(|| anyhow!(
                    "wrong webhook path used, expected /hook/{}",
                    cf.kind,
                ))
        };

        let forge: &config::Forge =
            correct_host_forges.clone()
            .find(|cf| check_kind(cf).is_ok())
            .ok_or_else(|| {
                let mut emsg = format!("no matching forge in config");
                for cf in correct_host_forges.clone() {
                    let wrong = check_kind(cf).expect_err("suddenly good?");
                    write!(emsg, "; forge host {:?}: {wrong}", cf.host)
                        .expect("write");
                }
                if correct_host_forges.clone().next().is_none() {
                    write!(emsg, "; no matching forge hosts")
                        .expect("write");
                }
                anyhow!("{}", emsg)
            })
            // TODO this is perhaps anthe actual permission denied variant,
            // log them at lower severity ?
            .map_err(WE::MisconfiguredWebhook)?;

        let _: IsAllowedClient = self.raw.client.allowed_by(&forge.allow)
            // TODO these are the actual permission denied variants,
            // log them at lower severity ?
            .await?;

        Ok(forge.host.clone())
    }

    fn check_tag_name(&self) -> Result<(), NotForUsReason> {
        let app_config = &self.globals.config.t2u;

        let (distro, version) = self.payload.tag_name.split('/')
            .collect_tuple()
            .ok_or_else(|| NFR::TagNameUnexpectedSyntax)?;
        (distro == app_config.distro).then(||())
            .ok_or_else(|| NFR::TagNameNotOurDistro)?;
        if !version.chars().all(
            |c| c.is_ascii_alphanumeric() || ".+-%_#".chars().contains(&c)
        ) {
            return Err(NFR::TagNameUnexpectedSyntax)
        }
        if version == "." || version == ".." {
            return Err(NFR::TagNameUnexpectedSyntax)
        }
        Ok(())
    }
}