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:
//!       * 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: ForgeKind> {
    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: ForgeKind> {
    globals: &'a Globals,
    forge: &'a F,
    payload: AnalysedPayload<F>,
    raw: &'a RawPayload,
}

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: ForgeKind> 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> {
        // We're going to log something, probably.
        // These IEFE's ensure that.

        let raw = (|| {

            self.payload

        })().inspect_err(|e| {
            debug!("rejected early: {e}");
        })?;

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

            let forge = F::default();
            raw.webhook_impl_inner(&forge, &mut log_info).await

        }.await.inspect_err(|e| {
            write_string!(log_info, "{e}");
            match e {
                WE::Throttled { .. } |
                WE::DisallowedClient { .. } |
                WE::PageNotFoundHere { .. } |
                // reported as Error when we generated it
                WE::InternalError { .. } => 
                    info!("rejected: {log_info}"),
                WE::MisconfiguredWebhook { .. } |
                WE::NotForUs { .. } => {
                    info!("ignored: {log_info}");
                    debug!("ignored: {log_info}, raw={:?}", &raw.raw);
                }
            }
        })
    }
}

impl RawPayload {
    async fn webhook_impl_inner<F: ForgeKind>(
        &self,
        forge: &F,
        log_info: &mut String,
    ) -> Result<String, WebError> {
        let payload = serde_json::from_value(self.raw.clone())
            .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_str().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)?;

        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 job_row = JobRow {
            jid: JobId::none(),
            data: validated_data,
            received: now,
            last_update: now,
            tag_data: None.into(),
            status: JobStatus::Noticed,
            info: format!("job received, tag not yet fetched"),
            processing: None.into(),
            duplicate_of: None,
        };

        let jid = db_transaction(TN::Update { 
            this_jid: None,
            tag_objectid: &job_row.data.tag_objectid,
        }, |dbt| {
            check_not_throttled(dbt)?;

            let jid = dbt.bsql_insert(bsql!(
                "INSERT INTO jobs " +~(job_row) ""
            )).into_internal("insert into jobs failed")?;

            Ok::<_, WebError>(jid)
        })??;

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

        info!(jid=%jid, now=?job_row.status, info=%job_row.info,
              "[{}] received", job_row.data.forge_host);

        Ok(msg)
    }
}

impl<'a, F: ForgeKind> 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)
                        .unwrap();
                }
                if correct_host_forges.clone().next().is_none() {
                    write!(emsg, "; no matching forge hosts")
                        .unwrap();
                }
                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(())
    }
}