this repo has no description
lewis.moe
1use crate::api::{ApiError, EmptyResponse};
2use crate::circuit_breaker::with_circuit_breaker;
3use crate::plc::{PlcClient, signing_key_to_did_key, validate_plc_operation};
4use crate::state::AppState;
5use axum::{
6 Json,
7 extract::State,
8 response::{IntoResponse, Response},
9};
10use k256::ecdsa::SigningKey;
11use serde::Deserialize;
12use serde_json::Value;
13use tracing::{error, info, warn};
14
15#[derive(Debug, Deserialize)]
16pub struct SubmitPlcOperationInput {
17 pub operation: Value,
18}
19
20pub async fn submit_plc_operation(
21 State(state): State<AppState>,
22 headers: axum::http::HeaderMap,
23 Json(input): Json<SubmitPlcOperationInput>,
24) -> Response {
25 let bearer = match crate::auth::extract_bearer_token_from_header(
26 headers.get("Authorization").and_then(|h| h.to_str().ok()),
27 ) {
28 Some(t) => t,
29 None => {
30 return ApiError::AuthenticationRequired.into_response();
31 }
32 };
33 let auth_user =
34 match crate::auth::validate_bearer_token_allow_deactivated(&state.db, &bearer).await {
35 Ok(user) => user,
36 Err(e) => {
37 return ApiError::from(e).into_response();
38 }
39 };
40 if let Err(e) = crate::auth::scope_check::check_identity_scope(
41 auth_user.is_oauth,
42 auth_user.scope.as_deref(),
43 crate::oauth::scopes::IdentityAttr::Wildcard,
44 ) {
45 return e;
46 }
47 let did = &auth_user.did;
48 if did.starts_with("did:web:") {
49 return ApiError::InvalidRequest(
50 "PLC operations are only valid for did:plc identities".into(),
51 )
52 .into_response();
53 }
54 if let Err(e) = validate_plc_operation(&input.operation) {
55 return ApiError::InvalidRequest(format!("Invalid operation: {}", e)).into_response();
56 }
57 let op = &input.operation;
58 let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
59 let public_url = format!("https://{}", hostname);
60 let user = match sqlx::query!("SELECT id, handle FROM users WHERE did = $1", did)
61 .fetch_optional(&state.db)
62 .await
63 {
64 Ok(Some(row)) => row,
65 _ => {
66 return ApiError::AccountNotFound.into_response();
67 }
68 };
69 let key_row = match sqlx::query!(
70 "SELECT key_bytes, encryption_version FROM user_keys WHERE user_id = $1",
71 user.id
72 )
73 .fetch_optional(&state.db)
74 .await
75 {
76 Ok(Some(row)) => row,
77 _ => {
78 return ApiError::InternalError(Some("User signing key not found".into())).into_response();
79 }
80 };
81 let key_bytes = match crate::config::decrypt_key(&key_row.key_bytes, key_row.encryption_version)
82 {
83 Ok(k) => k,
84 Err(e) => {
85 error!("Failed to decrypt user key: {}", e);
86 return ApiError::InternalError(None).into_response();
87 }
88 };
89 let signing_key = match SigningKey::from_slice(&key_bytes) {
90 Ok(k) => k,
91 Err(e) => {
92 error!("Failed to create signing key: {:?}", e);
93 return ApiError::InternalError(None).into_response();
94 }
95 };
96 let user_did_key = signing_key_to_did_key(&signing_key);
97 let server_rotation_key =
98 std::env::var("PLC_ROTATION_KEY").unwrap_or_else(|_| user_did_key.clone());
99 if let Some(rotation_keys) = op.get("rotationKeys").and_then(|v| v.as_array()) {
100 let has_server_key = rotation_keys
101 .iter()
102 .any(|k| k.as_str() == Some(&server_rotation_key));
103 if !has_server_key {
104 return ApiError::InvalidRequest(
105 "Rotation keys do not include server's rotation key".into(),
106 )
107 .into_response();
108 }
109 }
110 if let Some(services) = op.get("services").and_then(|v| v.as_object())
111 && let Some(pds) = services.get("atproto_pds").and_then(|v| v.as_object())
112 {
113 let service_type = pds.get("type").and_then(|v| v.as_str());
114 let endpoint = pds.get("endpoint").and_then(|v| v.as_str());
115 if service_type != Some("AtprotoPersonalDataServer") {
116 return ApiError::InvalidRequest("Incorrect type on atproto_pds service".into())
117 .into_response();
118 }
119 if endpoint != Some(&public_url) {
120 return ApiError::InvalidRequest("Incorrect endpoint on atproto_pds service".into())
121 .into_response();
122 }
123 }
124 if let Some(verification_methods) = op.get("verificationMethods").and_then(|v| v.as_object())
125 && let Some(atproto_key) = verification_methods.get("atproto").and_then(|v| v.as_str())
126 && atproto_key != user_did_key
127 {
128 return ApiError::InvalidRequest("Incorrect signing key in verificationMethods".into())
129 .into_response();
130 }
131 if let Some(also_known_as) = (!user.handle.is_empty())
132 .then(|| op.get("alsoKnownAs").and_then(|v| v.as_array()))
133 .flatten()
134 {
135 let expected_handle = format!("at://{}", user.handle);
136 let first_aka = also_known_as.first().and_then(|v| v.as_str());
137 if first_aka != Some(&expected_handle) {
138 return ApiError::InvalidRequest("Incorrect handle in alsoKnownAs".into())
139 .into_response();
140 }
141 }
142 let plc_client = PlcClient::with_cache(None, Some(state.cache.clone()));
143 let operation_clone = input.operation.clone();
144 let did_clone = did.clone();
145 if let Err(e) = with_circuit_breaker(&state.circuit_breakers.plc_directory, || async {
146 plc_client
147 .send_operation(&did_clone, &operation_clone)
148 .await
149 })
150 .await
151 {
152 return ApiError::from(e).into_response();
153 }
154 match sqlx::query!(
155 "INSERT INTO repo_seq (did, event_type, handle) VALUES ($1, 'identity', $2) RETURNING seq",
156 did,
157 user.handle
158 )
159 .fetch_one(&state.db)
160 .await
161 {
162 Ok(row) => {
163 if let Err(e) = sqlx::query(&format!("NOTIFY repo_updates, '{}'", row.seq))
164 .execute(&state.db)
165 .await
166 {
167 warn!("Failed to notify identity event: {:?}", e);
168 }
169 }
170 Err(e) => {
171 warn!("Failed to sequence identity event: {:?}", e);
172 }
173 }
174 let _ = state.cache.delete(&format!("handle:{}", user.handle)).await;
175 if state.did_resolver.refresh_did(did).await.is_none() {
176 warn!(did = %did, "Failed to refresh DID cache after PLC update");
177 }
178 info!(did = %did, "PLC operation submitted successfully");
179 EmptyResponse::ok().into_response()
180}