cratefield_import_supabase/
session.rs

1//! The read-only session every inspection read goes through.
2//!
3//! Read-only is enforced twice, as ADR 0026 (Decision 3) requires, and then
4//! checked:
5//!
6//! 1. the session's `default_transaction_read_only` is set on at startup and
7//!    again with `SET` once connected, so a statement outside the
8//!    transaction cannot write either;
9//! 2. every read runs inside one `REPEATABLE READ READ ONLY` transaction,
10//!    so all of it sees one snapshot, and the transaction is rolled back,
11//!    never committed;
12//! 3. before the rollback, `txid_current_if_assigned()` must still be null:
13//!    a write — even one refused — would have assigned a transaction id.
14//!
15//! The third guard is ours; the role is the user's. `docs/import/supabase.md`
16//! shows how to create a role that cannot write at all, and the report says
17//! whether the role used could ([`crate::report::ReadOnlyEvidence`]).
18
19use sqlx::Connection as _;
20use sqlx::postgres::{PgConnectOptions, PgConnection};
21use std::str::FromStr as _;
22
23use crate::secret::Secret;
24
25/// Why an inspection stopped. The exit code is non-zero for every one of
26/// them; a blocker is never one of them (it is part of the report).
27#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
28pub enum InspectError {
29    /// An input was refused before anything was contacted.
30    #[error("{0}")]
31    InvalidInput(String),
32    /// The database could not be reached or refused the login.
33    #[error("could not connect to the database: {0}")]
34    Connect(String),
35    /// The role lacks a privilege a read needs.
36    #[error("permission denied: {0}")]
37    Permission(String),
38    /// A read failed for another reason.
39    #[error("a catalog read failed: {0}")]
40    Query(String),
41    /// A transaction id was assigned during the inspection, which only a
42    /// write does. Should never happen; the inspection is abandoned.
43    #[error("a write was attempted during the inspection; nothing was committed")]
44    WriteDetected,
45    /// The Management API refused the token.
46    #[error("{0}")]
47    ManagementUnauthorized(String),
48}
49
50impl InspectError {
51    /// The error is about reaching or being allowed into a source — the
52    /// class the CLI's exit code reports.
53    #[must_use]
54    pub fn is_access(&self) -> bool {
55        matches!(
56            self,
57            Self::Connect(_) | Self::Permission(_) | Self::ManagementUnauthorized(_)
58        )
59    }
60}
61
62/// Strips everything a database error might echo back that the caller
63/// gave in secret: the URL and its password, then `scrub_text` for the
64/// rest.
65pub(crate) fn redact(message: &str, url: &Secret) -> String {
66    let mut message = message.replace(url.expose(), "[redacted]");
67    if let Ok(parsed) = url::Url::parse(url.expose())
68        && let Some(password) = parsed.password()
69        && !password.is_empty()
70    {
71        message = message.replace(password, "[redacted]");
72        let decoded = percent_decode(password);
73        if !decoded.is_empty() {
74            message = message.replace(&decoded, "[redacted]");
75        }
76    }
77    cratefield_core::scrub_text(&message)
78}
79
80/// `%XX` decoding, enough to recognise a URL-encoded password in plain
81/// text.
82fn percent_decode(text: &str) -> String {
83    let bytes = text.as_bytes();
84    let mut out = Vec::with_capacity(bytes.len());
85    let mut index = 0;
86    while index < bytes.len() {
87        if bytes[index] == b'%'
88            && index + 2 < bytes.len()
89            && text.is_char_boundary(index + 3)
90            && let Ok(byte) = u8::from_str_radix(&text[index + 1..index + 3], 16)
91        {
92            out.push(byte);
93            index += 3;
94        } else {
95            out.push(bytes[index]);
96            index += 1;
97        }
98    }
99    String::from_utf8_lossy(&out).into_owned()
100}
101
102/// Maps a sqlx error to the class the exit code and the message need.
103pub(crate) fn classify(error: &sqlx::Error, url: &Secret) -> InspectError {
104    let text = redact(&error.to_string(), url);
105    match error {
106        sqlx::Error::Database(db) => match db.code().as_deref() {
107            Some("42501") => InspectError::Permission(text),
108            Some(code) if code.starts_with("28") || code.starts_with("08") || code == "3D000" => {
109                InspectError::Connect(text)
110            }
111            _ => InspectError::Query(text),
112        },
113        sqlx::Error::Io(_)
114        | sqlx::Error::Tls(_)
115        | sqlx::Error::Configuration(_)
116        | sqlx::Error::Protocol(_)
117        | sqlx::Error::PoolTimedOut
118        | sqlx::Error::PoolClosed => InspectError::Connect(text),
119        _ => InspectError::Query(text),
120    }
121}
122
123/// One connection, in one read-only transaction.
124pub struct ReadOnlySession {
125    pub(crate) conn: PgConnection,
126    pub(crate) url: Secret,
127}
128
129impl ReadOnlySession {
130    /// Connects to `url` read-only and opens the `REPEATABLE READ READ
131    /// ONLY` transaction.
132    ///
133    /// # Errors
134    ///
135    /// [`InspectError::Connect`] when the database cannot be reached or
136    /// refuses the login; [`InspectError::InvalidInput`] when the URL does
137    /// not parse. Neither message quotes the URL.
138    pub async fn open(url: &Secret) -> Result<Self, InspectError> {
139        let options = PgConnectOptions::from_str(url.expose())
140            .map_err(|_| {
141                InspectError::InvalidInput(
142                    "the database URL is not a valid Postgres connection string".to_owned(),
143                )
144            })?
145            .application_name("cratefield-import-supabase")
146            .options([
147                ("default_transaction_read_only", "on"),
148                ("statement_timeout", "120000"),
149            ]);
150        let mut conn = PgConnection::connect_with(&options)
151            .await
152            .map_err(|error| classify(&error, url))?;
153        for statement in [
154            // Again, in case a pooler dropped the startup options.
155            "SET default_transaction_read_only = on",
156            "BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY",
157        ] {
158            sqlx::raw_sql(statement)
159                .execute(&mut conn)
160                .await
161                .map_err(|error| classify(&error, url))?;
162        }
163        Ok(Self {
164            conn,
165            url: url.clone(),
166        })
167    }
168
169    pub(crate) fn error(&self, error: &sqlx::Error) -> InspectError {
170        classify(error, &self.url)
171    }
172
173    /// Runs `sql` inside the session's transaction. For tests: the guards
174    /// have to refuse a write, and this is how one is attempted.
175    ///
176    /// # Errors
177    ///
178    /// The database's refusal, classified.
179    #[doc(hidden)]
180    pub async fn execute_for_test(&mut self, sql: &str) -> Result<(), InspectError> {
181        sqlx::raw_sql(sql)
182            .execute(&mut self.conn)
183            .await
184            .map(|_| ())
185            .map_err(|error| self.error(&error))
186    }
187
188    /// Asserts no transaction id was assigned, then rolls back.
189    ///
190    /// # Errors
191    ///
192    /// [`InspectError::WriteDetected`] when one was.
193    pub(crate) async fn finish(mut self) -> Result<bool, InspectError> {
194        let assigned: Option<String> =
195            sqlx::query_scalar("SELECT txid_current_if_assigned()::text")
196                .fetch_one(&mut self.conn)
197                .await
198                .map_err(|error| self.error(&error))?;
199        // Rolled back either way; a failed rollback still ends the session
200        // when the connection drops, and nothing was ever committed.
201        let _ = sqlx::raw_sql("ROLLBACK").execute(&mut self.conn).await;
202        let _ = self.conn.close().await;
203        if assigned.is_some() {
204            return Err(InspectError::WriteDetected);
205        }
206        Ok(true)
207    }
208}
cratefield_import_supabase/session.rs

cratefield_import_supabase/
session.rs
