feat: add subject access logs and appointment references

crates/registry-breg/src/api/access_log.rs

@@ -0,0 +1,148 @@
+// SPDX-License-Identifier: Apache-2.0
+
+use super::*;
+
+pub(super) fn routes(service: &HttpService) -> Router<Arc<HttpService>> {
+    let mut app = Router::new();
+    for route in &service.registry.routes().routes {
+        if route.operation != Operation::Get || read_path_for_route(service, route).is_some() {
+            continue;
+        }
+        if service
+            .registry
+            .entities()
+            .get(&route.entity_id)
+            .is_some_and(|entity| entity.access_log.is_some())
+        {
+            app = app.route(
+                &format!("{}/access-log", route.path),
+                get(read).layer(Extension(route.clone())),
+            );
+        }
+    }
+    app
+}
+
+async fn read(
+    State(service): State<Arc<HttpService>>,
+    Extension(route): Extension<CompiledRoute>,
+    Extension(correlation): Extension<RequestCorrelation>,
+    claims: Option<Extension<VerifiedRequestClaims>>,
+    RawQuery(raw_query): RawQuery,
+    headers: HeaderMap,
+    Path(path): Path<HashMap<String, String>>,
+) -> Response {
+    let claims = claims
+        .map(|Extension(value)| value)
+        .unwrap_or_else(VerifiedRequestClaims::anonymous);
+    let parsed = parse_options(raw_query.as_deref());
+    let (options, cursor, limit) = match parsed {
+        Ok(value) => value,
+        Err(()) => {
+            return audited_known_read_refusal(
+                &service,
+                &route,
+                &claims,
+                path.get("record_id"),
+                invalid_query(),
+                &correlation,
+            )
+            .await
+        }
+    };
+    let Some(surface) = authorize_route(&service, &route, &claims, &options) else {
+        return audited_read_concealment(
+            &service,
+            &route,
+            &options,
+            &claims,
+            path.get("record_id"),
+            &correlation,
+        )
+        .await;
+    };
+    let record_id = path.get("record_id");
+    if claims.principal().is_none()
+        || record_id.is_none_or(|id| !valid_canonical_record_uuid(id))
+        || headers.contains_key(crate::subject_access_log::REQUESTER_HEADER)
+        || headers.contains_key(crate::subject_access_log::PURPOSE_HEADER)
+    {
+        return audited_read_refusal(
+            &service,
+            &route,
+            &surface,
+            record_id,
+            concealed(),
+            &correlation,
+        )
+        .await;
+    }
+    let request = RecordReadRequest {
+        entity_id: route.entity_id.clone(),
+        operation_id: route.id.clone(),
+        method: route.method,
+        context: surface.context,
+        selected_fields: surface.readable_fields.clone(),
+        representation: CursorRepresentation::Json,
+        adapter: CursorAdapter::Native,
+        adapter_origin: None,
+        geojson_next_link_prefix: None,
+        kind: RecordReadKind::Get {
+            id: record_id.expect("validated record id").clone(),
+        },
+        maximum_records: 1,
+        request_history_after_proposal_version: None,
+        correlation: correlation.clone(),
+    };
+    match service.records.access_log(request, cursor, limit).await {
+        Ok(Some(held)) => {
+            let mut response = (
+                [(CONTENT_TYPE, HeaderValue::from_static("application/json"))],
+                held.body().to_vec(),
+            )
+                .into_response();
+            response
+                .headers_mut()
+                .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
+            response
+        }
+        Ok(None) => concealed(),
+        Err(ReadServiceError::CursorInvalid) => cursor_invalid(),
+        Err(_) => unavailable(),
+    }
+}
+
+fn parse_options(raw: Option<&str>) -> Result<(QueryOptions, Option<String>, u16), ()> {
+    let mut profile = None;
+    let mut cursor = None;
+    let mut limit = None;
+    if raw.is_some_and(|value| value.len() > 2048) {
+        return Err(());
+    }
+    for pair in raw
+        .unwrap_or_default()
+        .split('&')
+        .filter(|pair| !pair.is_empty())
+    {
+        let (name, value) = pair.split_once('=').ok_or(())?;
+        let name = percent_decode(name).map_err(|_| ())?;
+        let value = percent_decode(value).map_err(|_| ())?;
+        match name.as_str() {
+            "accessProfile" if profile.is_none() => profile = Some(value),
+            "cursor" if cursor.is_none() && valid_canonical_record_uuid(&value) => {
+                cursor = Some(value)
+            }
+            "limit" if limit.is_none() => {
+                let parsed: u16 = value.parse().map_err(|_| ())?;
+                if !(1..=100).contains(&parsed) || parsed.to_string() != value {
+                    return Err(());
+                }
+                limit = Some(parsed);
+            }
+            _ => return Err(()),
+        }
+    }
+    let mut options = QueryOptions::default();
+    options.parsed.access_profile = profile;
+    Ok((options, cursor, limit.unwrap_or(50)))
+}

crates/registry-breg/src/api/attachments.rs

@@ -226,6 +226,7 @@ async fn download(
     claims: Option<Extension<VerifiedRequestClaims>>,
     RawQuery(raw_query): RawQuery,
     Path(path): Path<HashMap<String, String>>,
+    headers: HeaderMap,
 ) -> Response {
     let claims = claims
         .map(|Extension(value)| value)
@@ -243,7 +244,7 @@ async fn download(
                 && record_id.is_some_and(|id| valid_canonical_record_uuid(id))
                 && surface.readable_fields.contains(&binding.slot)
         });
-    let Some(surface) = surface else {
+    let Some(mut surface) = surface else {
         return audited_known_read_refusal(
             &service,
             &attachment_route,
@@ -254,6 +255,21 @@ async fn download(
         )
         .await;
     };
+    if surface
+        .context
+        .bind_access_attribution(surface.response_entity, &headers)
+        .is_err()
+    {
+        return audited_known_read_refusal(
+            &service,
+            &attachment_route,
+            &claims,
+            record_id,
+            concealed(),
+            &correlation,
+        )
+        .await;
+    }
     let (_, version) = parsed.expect("authorized parsed attachment query");
     let request = RecordReadRequest {
         entity_id: route.entity_id.clone(),

crates/registry-breg/src/api/context.rs

@@ -333,6 +333,8 @@ impl fmt::Debug for VerifiedRowBoundary {
 pub struct AuthorizedRequestContext {
     principal: Option<String>,
     purpose: Option<String>,
+    requester_client: Option<String>,
+    forwarded_access_attribution: Option<crate::subject_access_log::ForwardedAccessAttribution>,
     selected_profile: String,
     row_boundaries: Vec<VerifiedRowBoundary>,
     request_actions: Vec<VerifiedRequestAction>,
@@ -354,6 +356,8 @@ impl AuthorizedRequestContext {
         Self {
             principal,
             purpose,
+            requester_client: None,
+            forwarded_access_attribution: None,
             selected_profile,
             row_boundaries,
             request_actions: Vec::new(),
@@ -388,8 +392,29 @@ impl AuthorizedRequestContext {
     pub(crate) fn with_grant_audit(mut self, claims: &VerifiedRequestClaims) -> Self {
         self.grant_audit = crate::audit::GrantAuditContext::from_claims(claims);
         self.human_identity = claims.human_identity().cloned();
+        self.requester_client = claims.requester_client().map(str::to_owned);
         self
     }
+
+    pub(crate) fn requester_client(&self) -> Option<&str> {
+        self.requester_client.as_deref()
+    }
+
+    pub(crate) fn forwarded_access_attribution(
+        &self,
+    ) -> Option<&crate::subject_access_log::ForwardedAccessAttribution> {
+        self.forwarded_access_attribution.as_ref()
+    }
+
+    pub(crate) fn bind_access_attribution(
+        &mut self,
+        entity: &crate::model::CompiledEntity,
+        headers: &axum::http::HeaderMap,
+    ) -> Result<(), ()> {
+        self.forwarded_access_attribution =
+            crate::subject_access_log::forwarded_attribution(entity, self, headers)?;
+        Ok(())
+    }
     pub(crate) fn grant_audit(&self) -> Option<&crate::audit::GrantAuditContext> {
         self.grant_audit.as_ref()
     }

crates/registry-breg/src/api/gis.rs

@@ -280,13 +280,30 @@ async fn items(
     claims: Option<Extension<VerifiedRequestClaims>>,
     RawQuery(raw_query): RawQuery,
     Path(collection): Path<String>,
+    headers: axum::http::HeaderMap,
 ) -> Response {
     let claims = claims
         .map(|Extension(value)| value)
         .unwrap_or_else(VerifiedRequestClaims::anonymous);
-    let Some(authorized) = authorize_gis_collection(&service, &claims, &collection) else {
+    let Some(mut authorized) = authorize_gis_collection(&service, &claims, &collection) else {
         return concealed();
     };
+    if authorized
+        .surface
+        .context
+        .bind_access_attribution(authorized.surface.response_entity, &headers)
+        .is_err()
+    {
+        return audited_read_refusal(
+            &service,
+            authorized.route,
+            &authorized.surface,
+            None,
+            concealed(),
+            &correlation,
+        )
+        .await;
+    }
     // The feature surface speaks its own parameter names, so a refusal here
     // stays unlocated rather than naming a native query parameter.
     let query = match parse_items_query(raw_query.as_deref()) {