mas_handlers/graphql/

mod.rs

// Copyright 2025, 2026 Element Creations Ltd.
// Copyright 2024, 2025 New Vector Ltd.
// Copyright 2022-2024 The Matrix.org Foundation C.I.C.
//
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
// Please see LICENSE files in the repository root for full details.

#![allow(
    clippy::module_name_repetitions,
    // async-graphql's `#[Object]` requires resolvers to be `async fn`, and its
    // `Enum` derive emits async trait impls without awaits
    clippy::unused_async_trait_impl,
)]

use std::{net::IpAddr, ops::Deref, sync::Arc};

use async_graphql::{
    EmptySubscription, InputObject,
    extensions::Tracing,
    http::MultipartOptions,
    parser::types::{DocumentOperations, OperationType},
};
use axum::{
    Extension, Json,
    body::Body,
    extract::{RawQuery, State as AxumState},
    http::StatusCode,
    response::{IntoResponse, Response},
};
use axum_extra::typed_header::TypedHeader;
use chrono::{DateTime, Utc};
use futures_util::TryStreamExt;
use headers::{Authorization, ContentType, HeaderValue, authorization::Bearer};
use hyper::header::CACHE_CONTROL;
use mas_axum_utils::{
    InternalError, RecordAsRequester, SessionInfo, SessionInfoExt, cookies::CookieJar, record_error,
};
use mas_data_model::{
    BoxClock, BoxRng, BrowserSession, Clock, Session, SiteConfig, SystemClock, User,
};
use mas_matrix::HomeserverConnection;
use mas_policy::{InstantiateError, Policy, PolicyFactory};
use mas_router::UrlBuilder;
use mas_storage::{BoxRepository, BoxRepositoryFactory, RepositoryError};
use opentelemetry_semantic_conventions::trace::{
    GRAPHQL_DOCUMENT, GRAPHQL_OPERATION_NAME, GRAPHQL_OPERATION_TYPE,
};
use rand::{SeedableRng, rng};
use rand_chacha::ChaChaRng;
use state::has_session_ended;
use tracing::{Instrument, info_span};
use ulid::Ulid;

mod model;
mod mutations;
mod query;
mod state;

pub use self::state::{BoxState, State};
use self::{
    model::{CreationEvent, Node},
    mutations::Mutation,
    query::Query,
};
use crate::{
    BoundActivityTracker, Limiter, RequesterFingerprint, impl_from_error_for_route,
    passwords::PasswordManager,
};

#[cfg(test)]
mod tests;

/// Extra parameters we get from the listener configuration, because they are
/// per-listener options. We pass them through request extensions.
#[derive(Debug, Clone)]
pub struct ExtraRouterParameters {
    pub undocumented_oauth2_access: bool,
}

struct GraphQLState {
    repository_factory: BoxRepositoryFactory,
    homeserver_connection: Arc<dyn HomeserverConnection>,
    policy_factory: Arc<PolicyFactory>,
    site_config: SiteConfig,
    password_manager: PasswordManager,
    url_builder: UrlBuilder,
    limiter: Limiter,
}

#[async_trait::async_trait]
impl state::State for GraphQLState {
    async fn repository(&self) -> Result<BoxRepository, RepositoryError> {
        self.repository_factory.create().await
    }

    async fn policy(&self) -> Result<Policy, InstantiateError> {
        self.policy_factory.instantiate().await
    }

    fn password_manager(&self) -> PasswordManager {
        self.password_manager.clone()
    }

    fn site_config(&self) -> &SiteConfig {
        &self.site_config
    }

    fn homeserver_connection(&self) -> &dyn HomeserverConnection {
        self.homeserver_connection.as_ref()
    }

    fn url_builder(&self) -> &UrlBuilder {
        &self.url_builder
    }

    fn limiter(&self) -> &Limiter {
        &self.limiter
    }

    fn clock(&self) -> BoxClock {
        let clock = SystemClock::default();
        Box::new(clock)
    }

    fn rng(&self) -> BoxRng {
        #[expect(clippy::disallowed_methods)]
        let mut rng = rng();

        let rng = ChaChaRng::from_rng(&mut rng);
        Box::new(rng)
    }
}

#[must_use]
pub fn schema(
    repository_factory: BoxRepositoryFactory,
    policy_factory: &Arc<PolicyFactory>,
    homeserver_connection: impl HomeserverConnection + 'static,
    site_config: SiteConfig,
    password_manager: PasswordManager,
    url_builder: UrlBuilder,
    limiter: Limiter,
) -> Schema {
    let state = GraphQLState {
        repository_factory,
        policy_factory: Arc::clone(policy_factory),
        homeserver_connection: Arc::new(homeserver_connection),
        site_config,
        password_manager,
        url_builder,
        limiter,
    };
    let state: BoxState = Box::new(state);

    schema_builder().extension(Tracing).data(state).finish()
}

fn span_and_operation_for_graphql_request(
    request: &mut async_graphql::Request,
) -> (tracing::Span, GraphQLOperation) {
    let span = info_span!(
        "GraphQL operation",
        "otel.name" = tracing::field::Empty,
        "otel.kind" = "server",
        { GRAPHQL_DOCUMENT } = request.query,
        { GRAPHQL_OPERATION_NAME } = tracing::field::Empty,
        { GRAPHQL_OPERATION_TYPE } = tracing::field::Empty,
    );

    let mut graphql_operation = GraphQLOperation {
        operation_type: None,
        operation_name: None,
    };

    // We need to clone the operation_name before parsing the query, else we're
    // going to have a borrow conflict between request.parsed_query() and
    // request.operation_name
    let operation_name = request.operation_name.clone();
    if let Ok(document) = request.parsed_query() {
        match (&document.operations, operation_name) {
            // A single anonymous operation, with no name requested: the
            // document defines no name for it, so we only record the type.
            (DocumentOperations::Single(operation), None) => {
                span.record("otel.name", format!("GraphQL {}", operation.node.ty));
                span.record(
                    GRAPHQL_OPERATION_TYPE,
                    tracing::field::display(operation.node.ty),
                );
                graphql_operation.operation_type = Some(operation.node.ty);
            }

            (DocumentOperations::Multiple(operations), Some(name)) => {
                if let Some((name, operation)) = operations.get_key_value(name.as_str()) {
                    span.record(
                        "otel.name",
                        format!("GraphQL {} {}", operation.node.ty, name),
                    );
                    span.record(
                        GRAPHQL_OPERATION_TYPE,
                        tracing::field::display(operation.node.ty),
                    );
                    span.record(GRAPHQL_OPERATION_NAME, tracing::field::display(name));
                    graphql_operation.operation_type = Some(operation.node.ty);
                    graphql_operation.operation_name = Some(name.to_string());
                }
            }

            (DocumentOperations::Multiple(operations), None) if operations.len() == 1 => {
                let mut iter = operations.iter();
                let (name, operation) = iter.next().unwrap();
                span.record(
                    "otel.name",
                    format!("GraphQL {} {}", operation.node.ty, name),
                );
                span.record(
                    GRAPHQL_OPERATION_TYPE,
                    tracing::field::display(operation.node.ty),
                );
                span.record(GRAPHQL_OPERATION_NAME, name.as_ref());
                graphql_operation.operation_type = Some(operation.node.ty);
                graphql_operation.operation_name = Some(name.to_string());
            }

            // Cases the executor rejects, so we don't record a misleading
            // operation: a single anonymous operation with a name requested, or
            // several named operations with no requested name (ambiguous).
            (DocumentOperations::Single(_), Some(_)) | (DocumentOperations::Multiple(_), None) => {}
        }
    }

    (span, graphql_operation)
}

/// The GraphQL operation being executed, attached to the response extensions so
/// the HTTP logging middleware can record it on the request log line.
#[derive(Clone, Debug)]
pub struct GraphQLOperation {
    /// The type of the operation: query, mutation or subscription.
    pub operation_type: Option<OperationType>,
    /// The name of the operation, as defined in the query document.
    pub operation_name: Option<String>,
}

#[derive(thiserror::Error, Debug)]
pub enum RouteError {
    #[error(transparent)]
    Internal(Box<dyn std::error::Error + Send + Sync + 'static>),

    #[error("Loading of some database objects failed")]
    LoadFailed,

    #[error("Invalid access token")]
    InvalidToken,

    #[error("Missing scope")]
    MissingScope,

    #[error(transparent)]
    ParseRequest(#[from] async_graphql::ParseRequestError),
}

impl_from_error_for_route!(mas_storage::RepositoryError);

impl IntoResponse for RouteError {
    fn into_response(self) -> Response {
        let sentry_event_id = record_error!(self, Self::Internal(_) | Self::LoadFailed);

        let response = match self {
            e @ (Self::Internal(_) | Self::LoadFailed) => {
                let error = async_graphql::Error::new_with_source(e);
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    Json(serde_json::json!({"errors": [error]})),
                )
                    .into_response()
            }

            Self::InvalidToken => {
                let error = async_graphql::Error::new("Invalid token");
                (
                    StatusCode::UNAUTHORIZED,
                    Json(serde_json::json!({"errors": [error]})),
                )
                    .into_response()
            }

            Self::MissingScope => {
                let error = async_graphql::Error::new("Missing urn:mas:graphql:* scope");
                (
                    StatusCode::UNAUTHORIZED,
                    Json(serde_json::json!({"errors": [error]})),
                )
                    .into_response()
            }

            Self::ParseRequest(e) => {
                let error = async_graphql::Error::new_with_source(e);
                (
                    StatusCode::BAD_REQUEST,
                    Json(serde_json::json!({"errors": [error]})),
                )
                    .into_response()
            }
        };

        (sentry_event_id, response).into_response()
    }
}

async fn get_requester(
    undocumented_oauth2_access: bool,
    clock: &impl Clock,
    activity_tracker: &BoundActivityTracker,
    mut repo: BoxRepository,
    session_info: &SessionInfo,
    user_agent: Option<String>,
    token: Option<&str>,
) -> Result<Requester, RouteError> {
    let entity = if let Some(token) = token {
        // If we haven't enabled undocumented_oauth2_access on the listener, we bail out
        if !undocumented_oauth2_access {
            return Err(RouteError::InvalidToken);
        }

        let token = repo
            .oauth2_access_token()
            .find_by_token(token)
            .await?
            .ok_or(RouteError::InvalidToken)?;

        let session = repo
            .oauth2_session()
            .lookup(token.session_id)
            .await?
            .ok_or(RouteError::LoadFailed)?;

        activity_tracker
            .record_oauth2_session(clock, &session)
            .await;

        // Load the user if there is one
        let user = if let Some(user_id) = session.user_id {
            let user = repo
                .user()
                .lookup(user_id)
                .await?
                .ok_or(RouteError::LoadFailed)?;
            Some(user)
        } else {
            None
        };

        // If there is a user for this session, check that it is not locked
        let user_valid = user.as_ref().is_none_or(User::is_valid);

        if !token.is_valid(clock.now()) || !session.is_valid() || !user_valid {
            return Err(RouteError::InvalidToken);
        }

        if !session.scope.contains("urn:mas:graphql:*") {
            return Err(RouteError::MissingScope);
        }

        if let Some(user) = &user {
            user.maybe_record_as_requester();
        }

        RequestingEntity::OAuth2Session(Box::new((session, user)))
    } else {
        let maybe_session = session_info.load_active_session(&mut repo).await?;

        if let Some(session) = maybe_session.as_ref() {
            activity_tracker
                .record_browser_session(clock, session)
                .await;
        }

        RequestingEntity::from(maybe_session)
    };

    let requester = Requester {
        entity,
        ip_address: activity_tracker.ip(),
        user_agent,
    };

    repo.cancel().await?;
    Ok(requester)
}

/// Log the errors of a GraphQL response which were caused by an internal
/// error, so that they get reported to Sentry.
///
/// Resolvers return internal errors through `?` on an [`anyhow::Error`] or a
/// [`mas_storage::RepositoryError`], which async-graphql keeps as the error
/// source. Other errors are user-facing and only logged by the tracing
/// extension, so a new internal error type reaching a resolver through `?`
/// has to be added here to be reported.
fn record_internal_errors(response: &async_graphql::Response) {
    for error in &response.errors {
        let Some(source) = error.source.as_deref() else {
            continue;
        };

        let source: &(dyn std::error::Error + 'static) =
            if let Some(err) = source.downcast_ref::<anyhow::Error>() {
                err.as_ref()
            } else if let Some(err) = source.downcast_ref::<mas_storage::RepositoryError>() {
                err
            } else {
                continue;
            };

        // Build a string of the path to where the error was, like "users.nodes.id",
        // leaving out the array indices in the path if any.
        let path = error
            .path
            .iter()
            .filter_map(|segment| match segment {
                async_graphql::PathSegment::Field(name) => Some(name.as_str()),
                async_graphql::PathSegment::Index(_) => None,
            })
            .collect::<Vec<_>>()
            .join(".");

        tracing::error!(
            error = source,
            tags.graphql.path = path,
            "Internal error while resolving a GraphQL request"
        );
    }
}

pub async fn post(
    AxumState(schema): AxumState<Schema>,
    Extension(ExtraRouterParameters {
        undocumented_oauth2_access,
    }): Extension<ExtraRouterParameters>,
    clock: BoxClock,
    repo: BoxRepository,
    activity_tracker: BoundActivityTracker,
    cookie_jar: CookieJar,
    content_type: Option<TypedHeader<ContentType>>,
    authorization: Option<TypedHeader<Authorization<Bearer>>>,
    user_agent: Option<TypedHeader<headers::UserAgent>>,
    body: Body,
) -> Result<impl IntoResponse, RouteError> {
    let body = body.into_data_stream();
    let token = authorization
        .as_ref()
        .map(|TypedHeader(Authorization(bearer))| bearer.token());
    let user_agent = user_agent.map(|TypedHeader(h)| h.to_string());
    let (session_info, mut cookie_jar) = cookie_jar.session_info();
    let requester = get_requester(
        undocumented_oauth2_access,
        &clock,
        &activity_tracker,
        repo,
        &session_info,
        user_agent,
        token,
    )
    .await?;

    let content_type = content_type.map(|TypedHeader(h)| h.to_string());

    let mut request = async_graphql::http::receive_body(
        content_type,
        body.map_err(std::io::Error::other).into_async_read(),
        MultipartOptions::default(),
    )
    .await?
    .data(requester); // XXX: this should probably return another error response?

    let (span, operation) = span_and_operation_for_graphql_request(&mut request);
    let mut response = schema.execute(request).instrument(span).await;
    record_internal_errors(&response);

    if has_session_ended(&mut response) {
        let session_info = session_info.mark_session_ended(clock.now());
        cookie_jar = cookie_jar.update_session_info(&session_info);
    }

    let cache_control = response
        .cache_control
        .value()
        .and_then(|v| HeaderValue::from_str(&v).ok())
        .map(|h| [(CACHE_CONTROL, h)]);

    let headers = response.http_headers.clone();

    Ok((
        headers,
        cache_control,
        cookie_jar,
        Extension(operation),
        Json(response),
    ))
}

pub async fn get(
    AxumState(schema): AxumState<Schema>,
    Extension(ExtraRouterParameters {
        undocumented_oauth2_access,
    }): Extension<ExtraRouterParameters>,
    clock: BoxClock,
    repo: BoxRepository,
    activity_tracker: BoundActivityTracker,
    cookie_jar: CookieJar,
    authorization: Option<TypedHeader<Authorization<Bearer>>>,
    user_agent: Option<TypedHeader<headers::UserAgent>>,
    RawQuery(query): RawQuery,
) -> Result<impl IntoResponse, InternalError> {
    let token = authorization
        .as_ref()
        .map(|TypedHeader(Authorization(bearer))| bearer.token());
    let user_agent = user_agent.map(|TypedHeader(h)| h.to_string());
    let (session_info, mut cookie_jar) = cookie_jar.session_info();
    let requester = get_requester(
        undocumented_oauth2_access,
        &clock,
        &activity_tracker,
        repo,
        &session_info,
        user_agent,
        token,
    )
    .await?;

    let mut request =
        async_graphql::http::parse_query_string(&query.unwrap_or_default())?.data(requester);

    let (span, operation) = span_and_operation_for_graphql_request(&mut request);
    let mut response = schema.execute(request).instrument(span).await;
    record_internal_errors(&response);

    if has_session_ended(&mut response) {
        let session_info = session_info.mark_session_ended(clock.now());
        cookie_jar = cookie_jar.update_session_info(&session_info);
    }

    let cache_control = response
        .cache_control
        .value()
        .and_then(|v| HeaderValue::from_str(&v).ok())
        .map(|h| [(CACHE_CONTROL, h)]);

    let headers = response.http_headers.clone();

    Ok((
        headers,
        cache_control,
        cookie_jar,
        Extension(operation),
        Json(response),
    ))
}

pub type Schema = async_graphql::Schema<Query, Mutation, EmptySubscription>;
pub type SchemaBuilder = async_graphql::SchemaBuilder<Query, Mutation, EmptySubscription>;

#[must_use]
pub fn schema_builder() -> SchemaBuilder {
    async_graphql::Schema::build(Query::new(), Mutation::new(), EmptySubscription)
        .register_output_type::<Node>()
        .register_output_type::<CreationEvent>()
}

pub struct Requester {
    entity: RequestingEntity,
    ip_address: Option<IpAddr>,
    user_agent: Option<String>,
}

impl Requester {
    pub fn fingerprint(&self) -> RequesterFingerprint {
        if let Some(ip) = self.ip_address {
            RequesterFingerprint::new(ip)
        } else {
            RequesterFingerprint::EMPTY
        }
    }

    pub fn for_policy(&self) -> mas_policy::Requester {
        mas_policy::Requester {
            ip_address: self.ip_address,
            user_agent: self.user_agent.clone(),
        }
    }
}

impl Deref for Requester {
    type Target = RequestingEntity;

    fn deref(&self) -> &Self::Target {
        &self.entity
    }
}

/// The identity of the requester.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub enum RequestingEntity {
    /// The requester presented no authentication information.
    #[default]
    Anonymous,

    /// The requester is a browser session, stored in a cookie.
    BrowserSession(Box<BrowserSession>),

    /// The requester is a `OAuth2` session, with an access token.
    OAuth2Session(Box<(Session, Option<User>)>),
}

trait OwnerId {
    fn owner_id(&self) -> Option<Ulid>;
}

impl OwnerId for User {
    fn owner_id(&self) -> Option<Ulid> {
        Some(self.id)
    }
}

impl OwnerId for BrowserSession {
    fn owner_id(&self) -> Option<Ulid> {
        Some(self.user.id)
    }
}

impl OwnerId for mas_data_model::UserEmail {
    fn owner_id(&self) -> Option<Ulid> {
        Some(self.user_id)
    }
}

impl OwnerId for Session {
    fn owner_id(&self) -> Option<Ulid> {
        self.user_id
    }
}

impl OwnerId for mas_data_model::CompatSession {
    fn owner_id(&self) -> Option<Ulid> {
        Some(self.user_id)
    }
}

impl OwnerId for mas_data_model::UpstreamOAuthLink {
    fn owner_id(&self) -> Option<Ulid> {
        self.user_id
    }
}

/// A dumb wrapper around a `Ulid` to implement `OwnerId` for it.
pub struct UserId(Ulid);

impl OwnerId for UserId {
    fn owner_id(&self) -> Option<Ulid> {
        Some(self.0)
    }
}

impl RequestingEntity {
    fn browser_session(&self) -> Option<&BrowserSession> {
        match self {
            Self::BrowserSession(session) => Some(session),
            Self::OAuth2Session(_) | Self::Anonymous => None,
        }
    }

    fn user(&self) -> Option<&User> {
        match self {
            Self::BrowserSession(session) => Some(&session.user),
            Self::OAuth2Session(tuple) => tuple.1.as_ref(),
            Self::Anonymous => None,
        }
    }

    fn oauth2_session(&self) -> Option<&Session> {
        match self {
            Self::OAuth2Session(tuple) => Some(&tuple.0),
            Self::BrowserSession(_) | Self::Anonymous => None,
        }
    }

    /// Returns true if the requester can access the resource.
    fn is_owner_or_admin(&self, resource: &impl OwnerId) -> bool {
        // If the requester is an admin, they can do anything.
        if self.is_admin() {
            return true;
        }

        // Otherwise, they must be the owner of the resource.
        let Some(owner_id) = resource.owner_id() else {
            return false;
        };

        let Some(user) = self.user() else {
            return false;
        };

        user.id == owner_id
    }

    fn is_admin(&self) -> bool {
        match self {
            Self::OAuth2Session(tuple) => {
                // TODO: is this the right scope?
                // This has to be in sync with the policy
                tuple.0.scope.contains("urn:mas:admin")
            }
            Self::BrowserSession(_) | Self::Anonymous => false,
        }
    }
}

impl From<BrowserSession> for RequestingEntity {
    fn from(session: BrowserSession) -> Self {
        Self::BrowserSession(Box::new(session))
    }
}

impl<T> From<Option<T>> for RequestingEntity
where
    T: Into<RequestingEntity>,
{
    fn from(session: Option<T>) -> Self {
        session.map(Into::into).unwrap_or_default()
    }
}

/// A filter for dates, with a lower bound and an upper bound
#[derive(InputObject, Default, Clone, Copy)]
pub struct DateFilter {
    /// The lower bound of the date range
    after: Option<DateTime<Utc>>,

    /// The upper bound of the date range
    before: Option<DateTime<Utc>>,
}