// SPDX-License-Identifier: AGPL-3.0-only // SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® module; #include #include #include #include module Crafter.Review:PgWire_impl; import :PgWire; import :Json; import :Text; import :Crypto; import :Platform; import Crafter.Network; import std; namespace Crafter::Review::PgWire { namespace { // ---- the statement scanner, on code points ---------------------------------- bool IdentChar(char32_t c) { return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '_' || c == '$' || (c >= 0x80 && c <= 0xFFFF); } bool TagStart(char32_t c) { return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || c == '_' || (c >= 0x80 && c <= 0xFFFF); } bool TagChar(char32_t c) { return TagStart(c) || (c >= '0' && c <= '9'); } bool At(std::u32string_view s, std::size_t i, std::u32string_view what) { return s.substr(i).starts_with(what); } // DOLLAR_TAG.match(s, i): $tag$ or $$, else empty. std::u32string_view DollarTag(std::u32string_view s, std::size_t i) { if (i >= s.size() || s[i] != '$') return {}; std::size_t j = i + 1; if (j < s.size() && TagStart(s[j])) { ++j; while (j < s.size() && TagChar(s[j])) ++j; } if (j < s.size() && s[j] == '$') return s.substr(i, j + 1 - i); return {}; } // The end of a `--` comment: Postgres ends it at \n or \r. std::size_t EndOfLine(std::u32string_view s, std::size_t i) { while (i < s.size() && s[i] != '\n' && s[i] != '\r') ++i; return i; } // The end of a `/* */` comment; they nest. std::size_t EndBlockComment(std::u32string_view s, std::size_t i) { std::int64_t depth = 0; while (i < s.size()) { if (At(s, i, U"/*")) { ++depth; i += 2; } else if (At(s, i, U"*/")) { --depth; i += 2; if (depth == 0) return i; } else { ++i; } } return s.size(); } std::size_t EndQuote(std::u32string_view s, std::size_t i, char32_t quote, bool backslash) { ++i; while (i < s.size()) { if (backslash && s[i] == '\\') { i += 2; continue; } if (s[i] == quote) { if (i + 1 < s.size() && s[i + 1] == quote) { i += 2; continue; } return i + 1; } ++i; } return s.size(); } // Past whitespace and comments from i. std::size_t Skip(std::u32string_view s, std::size_t i) { while (i < s.size()) { if (Dyn::IsSpace(s[i])) { ++i; } else if (At(s, i, U"--")) { i = EndOfLine(s, i); } else if (At(s, i, U"/*")) { i = EndBlockComment(s, i); } else { break; } } return i; } // An E'...' string: a backslash escapes in it. bool EscapeString(std::u32string_view s, std::size_t i) { return i > 0 && (s[i - 1] == 'e' || s[i - 1] == 'E') && (i < 2 || !IdentChar(s[i - 2])); } constexpr std::array RoutineWords = {"create", "or", "replace", "function", "procedure"}; // Does the statement start CREATE [OR REPLACE] FUNCTION|PROCEDURE // (psql's test, on its first four words)? bool Routine(const std::vector& marks) { auto mark = [&](std::size_t i) { return i < marks.size() ? std::string_view(marks[i]) : std::string_view(); }; auto routine = [](std::string_view w) { return w == "function" || w == "procedure"; }; return mark(0) == "create" && (routine(mark(1)) || (mark(1) == "or" && mark(2) == "replace" && routine(mark(3)))); } // ---- the wire --------------------------------------------------------------------- // A failure of the login exchange or of an answer: its kind and its // text, for "the login exchange broke (KeyError: 'r')". class Broke : public std::runtime_error { public: Broke(std::string type, const std::string& message) : std::runtime_error(message), type(std::move(type)) {} std::string type; }; // A socket error, as its message prints: "[Errno 104] Connection reset // by peer", or "timed out". class SocketError : public std::runtime_error { public: SocketError(const std::string& text, bool timeout) : std::runtime_error(text), timeout(timeout) {} bool timeout; }; std::string ConnectErrorText(const std::exception& error) { if (dynamic_cast(&error)) return "timed out"; if (const auto* failure = dynamic_cast(&error)) { if (failure->kind == SocketErrorKind::Resolver) return std::format("[Errno {}] {}", failure->code, gai_strerror(failure->code)); return Dyn::ErrnoText(failure->code); } return error.what(); } std::string Be32(std::uint32_t value) { std::string out(4, '\0'); for (std::int32_t i = 0; i < 4; ++i) out[static_cast(i)] = static_cast((value >> (24 - 8 * i)) & 0xFF); return out; } std::string Be16(std::uint16_t value) { return {static_cast(value >> 8), static_cast(value & 0xFF)}; } std::string CString(std::string_view text) { return std::format("{}{}", text, std::string(1, '\0')); } std::string Message(char type, std::string_view payload = {}) { return std::format("{}{}{}", type, Be32(static_cast(payload.size() + 4)), payload); } std::uint32_t Unpack32(std::string_view bytes, std::size_t at) { if (at + 4 > bytes.size()) throw Broke("error", "unpack requires a buffer of 4 bytes"); std::uint32_t value = 0; for (std::size_t i = 0; i < 4; ++i) value = (value << 8) | static_cast(bytes[at + i]); return value; } std::uint16_t Unpack16(std::string_view bytes, std::size_t at) { if (at + 2 > bytes.size()) throw Broke("error", "unpack requires a buffer of 2 bytes"); return static_cast((static_cast(bytes[at]) << 8) | static_cast(bytes[at + 1])); } // bytes.decode(): strict. std::string Decode(std::string_view bytes) { if (const std::optional error = Dyn::Utf8DecodeError(bytes)) throw Broke("UnicodeDecodeError", *error); return std::string(bytes); } // ErrorResponse / NoticeResponse fields. Fields FieldsOf(std::string_view payload) { Fields out; std::size_t i = 0; while (i < payload.size() && payload[i] != '\0') { const std::size_t j = payload.find('\0', i + 1); if (j == std::string_view::npos) throw Broke("ValueError", "subsection not found"); out[payload[i]] = Dyn::DecodeReplace(payload.substr(i + 1, j - i - 1)); i = j + 1; } return out; } std::string Get(const Fields& fields, char key, std::string_view fallback = {}) { const auto found = fields.find(key); return found == fields.end() ? std::string(fallback) : found->second; } // `f.get(a) or f.get(b) or fallback` std::string FirstOf(const Fields& fields, char a, char b, std::string_view fallback) { if (const std::string first = Get(fields, a); !first.empty()) return first; if (const std::string second = Get(fields, b); !second.empty()) return second; return std::string(fallback); } // A SCRAM message's attributes: comma-separated, each cut at its first // '='. std::map Attributes(std::string_view text) { std::map out; std::size_t n = 0; for (const std::string& item : Dyn::Split(text, ",")) { const std::size_t equals = item.find('='); if (equals == std::string::npos) throw Broke("ValueError", std::format("dictionary update sequence element #{} has length 1; 2 is required", n)); out[item.substr(0, equals)] = item.substr(equals + 1); ++n; } return out; } const std::string& Attribute(const std::map& attributes, const std::string& key) { const auto found = attributes.find(key); if (found == attributes.end()) throw Broke("KeyError", Dyn::Repr(key)); return found->second; } std::string Base64(std::string_view text) { try { return Base64Decode(text); } catch (const std::invalid_argument& error) { throw Broke("Error", error.what()); } } constexpr std::array RetrySqlStates = {"53300", "57P03"}; } // ---- the URL --------------------------------------------------------------------------- Url ParseUrl(std::string_view text) { const Dyn::SplitUrl split = Dyn::UrlSplit(text); Url url; url.scheme = split.scheme; url.host = Dyn::Lower(split.Hostname().value_or("")); const std::optional port = split.Port(); url.port = port && *port != 0 ? *port : 5432; const std::optional user = split.Username(); url.user = Dyn::Unquote(user && !user->empty() ? *user : "postgres"); url.password = Dyn::Unquote(split.Password().value_or("")); const std::string path = split.path.empty() ? "/" : split.path; const std::string database = Dyn::LStrip(path, "/"); url.database = Dyn::Unquote(database.empty() ? "postgres" : database); return url; } std::optional CheckUrl(const Json& url) { if (!url.is_string() || url.get_ref().empty()) return "no database URL"; Url parsed; try { parsed = ParseUrl(url.get_ref()); } catch (const std::invalid_argument& error) { return std::format("not a URL ({})", error.what()); } if (parsed.scheme != "postgres" && parsed.scheme != "postgresql") return "not a postgres:// or postgresql:// URL"; if (!std::ranges::contains(Loopback, parsed.host)) { const std::vector names(Loopback.begin(), Loopback.end()); return std::format("host {} is not this machine ({}): the database under test is a local stack, never a hosted one", parsed.host.empty() ? "(none)" : parsed.host, Dyn::Join(names, ", ")); } return std::nullopt; } std::string Redact(std::string_view url) { try { const Dyn::SplitUrl split = Dyn::UrlSplit(url); if (!split.Password()) return std::string(url); const std::string host = split.netloc.substr(split.netloc.rfind('@') + 1); const std::string netloc = std::format("{}:***@{}", split.Username().value_or(""), host); return Dyn::UrlUnsplit(split.scheme, netloc, split.path, split.query, split.fragment); } catch (const std::invalid_argument&) { return "(unparsable URL)"; } } std::string Where(std::string_view url) { const Url parsed = ParseUrl(url); return std::format("{} on {}:{} as {}", parsed.database, parsed.host, parsed.port, parsed.user); } // ---- statements ---------------------------------------------------------------------- std::vector SplitStatements(std::string_view text) { const std::u32string sql = Dyn::Decode(text); const std::u32string_view s(sql); const std::size_t n = s.size(); std::vector pieces; std::size_t start = 0; std::size_t i = 0; std::int64_t depth = 0; // psql's identifiers[], identifier_count and begin_depth, per // statement. std::vector marks; std::int64_t words = 0; std::int64_t block = 0; while (i < n) { const char32_t c = s[i]; const bool letter = (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || c == '_' || c > 127; if (letter && !(i > 0 && IdentChar(s[i - 1])) && !At(s, i, U"e'") && !At(s, i, U"E'")) { std::size_t j = i + 1; while (j < n && IdentChar(s[j])) ++j; const std::string word = Dyn::Lower(Dyn::Encode(s.substr(i, j - i))); if (words < 4) marks.push_back(std::ranges::contains(RoutineWords, word) ? word : std::string()); ++words; if (depth == 0 && Routine(marks)) { if (word == "begin") { ++block; } else if (word == "case" && block >= 1) { ++block; } else if (word == "end" && block > 0) { --block; } } i = j; continue; } if (At(s, i, U"--")) { i = EndOfLine(s, i); continue; } if (At(s, i, U"/*")) { i = EndBlockComment(s, i); continue; } if (c == '\'') { i = EndQuote(s, i, '\'', EscapeString(s, i)); continue; } if (c == '"') { i = EndQuote(s, i, '"', false); continue; } if (c == '$' && !(i > 0 && IdentChar(s[i - 1]))) { const std::u32string_view tag = DollarTag(s, i); if (!tag.empty()) { const std::size_t close = s.find(tag, i + tag.size()); i = close == std::u32string_view::npos ? n : close + tag.size(); continue; } } if (c == '(') { ++depth; } else if (c == ')') { depth = std::max(0, depth - 1); } else if (c == ';' && depth == 0 && block == 0) { pieces.push_back(s.substr(start, i - start)); start = i + 1; marks.clear(); words = 0; block = 0; } ++i; } pieces.push_back(s.substr(start)); std::vector out; for (const std::u32string_view piece : pieces) { if (Skip(piece, 0) < piece.size()) out.push_back(Dyn::Strip(Dyn::Encode(piece))); } return out; } std::string StripComments(std::string_view text) { const std::u32string sql = Dyn::Decode(text); const std::u32string_view s(sql); std::u32string out; std::size_t i = 0; while (i < s.size()) { if (At(s, i, U"--")) { i = EndOfLine(s, i); continue; } if (At(s, i, U"/*")) { i = EndBlockComment(s, i); continue; } const char32_t c = s[i]; std::optional end; if (c == '\'') { end = EndQuote(s, i, '\'', EscapeString(s, i)); } else if (c == '"') { end = EndQuote(s, i, '"', false); } else if (c == '$' && !(i > 0 && IdentChar(s[i - 1]))) { const std::u32string_view tag = DollarTag(s, i); if (!tag.empty()) { const std::size_t close = s.find(tag, i + tag.size()); end = close == std::u32string_view::npos ? s.size() : close + tag.size(); } } if (!end) { out.push_back(c); ++i; } else { out.append(s.substr(i, *end - i)); i = *end; } } return Dyn::Encode(out); } std::string Fingerprint(std::string_view sql) { std::u32string kept; for (const char32_t c : Dyn::Decode(StripComments(sql))) { if (!Dyn::IsSpace(c) && c != ';') kept.push_back(c); } return Sha256Hex(Dyn::Encode(kept)); } std::vector FirstWords(std::string_view text, std::size_t k) { const std::u32string statement = Dyn::Decode(text); const std::u32string_view s(statement); std::vector words; std::size_t i = 0; auto wordStart = [](char32_t c) { return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || c == '_'; }; auto wordChar = [&](char32_t c) { return wordStart(c) || (c >= '0' && c <= '9') || c == '$'; }; while (words.size() < k) { i = Skip(s, i); while (i < s.size() && s[i] == '(' && words.empty()) i = Skip(s, i + 1); if (i >= s.size() || !wordStart(s[i])) break; std::size_t j = i + 1; while (j < s.size() && wordChar(s[j])) ++j; std::string word = Dyn::Encode(s.substr(i, j - i)); for (char& ch : word) { if (ch >= 'A' && ch <= 'Z') ch = static_cast(ch - 'A' + 'a'); } words.push_back(std::move(word)); i = j; } return words; } std::optional Refused(std::string_view statement) { const std::vector w = FirstWords(statement, 4); const std::string k = w.empty() ? std::string() : w[0]; std::span rest = w.empty() ? std::span() : std::span(w).subspan(1); if (k == "rollback" && !rest.empty() && (rest[0] == "work" || rest[0] == "transaction")) rest = rest.subspan(1); const bool to = !rest.empty() && rest[0] == "to"; if (k == "begin" || k == "start" || k == "commit" || k == "end" || k == "abort" || (k == "rollback" && !to) || (k == "prepare" && !rest.empty() && rest[0] == "transaction")) { return std::format("{} - the helper owns the transaction and rolls it back at the end of the call (SAVEPOINT and ROLLBACK TO are fine)", Dyn::Upper(k)); } if (k == "copy") return "COPY - it reads or writes files, programs or the client outside the transaction; use SELECT / INSERT"; return std::nullopt; } std::string Headline(std::string_view statement, std::size_t width) { const std::string line = Dyn::Join(Dyn::Split(statement), " "); if (Dyn::Length(line) <= width) return line; return std::format("{}…", Dyn::Slice(line, 0, static_cast(width) - 1)); } // ---- the connection ---------------------------------------------------------------------- SqlError::SqlError(Fields f) : std::runtime_error(Get(f, 'M', "error")), fields(std::move(f)) {} struct Connection::State { Url url; std::unique_ptr socket; std::string buffer; std::size_t position = 0; char status = '\0'; std::vector notices; std::map parameters; // `count` bytes, or fewer when the server closed first. std::string ReadExact(std::size_t count) { while (buffer.size() - position < count) { if (position > 0 && position == buffer.size()) { buffer.clear(); position = 0; } std::array chunk {}; try { const std::uint32_t got = socket->RecieveSync(static_cast(chunk.size()), chunk.data()); buffer.append(chunk.data(), got); } catch (const SocketClosedException&) { break; } catch (const SocketTimeoutException&) { throw SocketError("timed out", true); } catch (const SocketErrorException& error) { throw SocketError(Dyn::ErrnoText(error.code), false); } } const std::size_t take = std::min(count, buffer.size() - position); std::string out = buffer.substr(position, take); position += take; return out; } std::pair Read() { const std::string header = ReadExact(5); if (header.size() < 5) throw DbError("the server closed the connection"); const std::uint32_t length = Unpack32(header, 1); const std::size_t bodyLength = length >= 4 ? length - 4 : 0; std::string body = ReadExact(bodyLength); if (body.size() < bodyLength) throw DbError("the server closed the connection"); return {header[0], std::move(body)}; } void Send(std::string_view data) { try { socket->Send(data.data(), static_cast(data.size())); } catch (const SocketErrorException& error) { throw DbError(std::format("the connection broke ({})", Dyn::ErrnoText(error.code))); } catch (const SocketClosedException&) { throw DbError(std::format("the connection broke ({})", Dyn::ErrnoText(32))); } } void Authenticate() { std::optional scram; while (true) { std::pair message; try { message = Read(); } catch (const SocketError& error) { throw DbError(std::format("no answer at login ({})", error.what())); } const auto& [type, body] = message; if (type == 'E') { const Fields fields = FieldsOf(body); throw DbError(Dyn::Strip(std::format("login refused: {} {}", Get(fields, 'C'), Get(fields, 'M')))); } if (type == 'R') { const std::uint32_t code = Unpack32(body, 0); if (code == 0) continue; if (code == 3) { Send(Message('p', CString(url.password))); } else if (code == 5) { const std::string inner = Md5Hex(url.password + url.user); const std::string salt = body.substr(4, 4); Send(Message('p', CString(std::format("md5{}", Md5Hex(inner + salt))))); } else if (code == 10) { Json mechanisms = Json::array(); bool scramOffered = false; for (const std::string& mechanism : Dyn::Split(std::string_view(body).substr(4), std::string_view("\0", 1))) { if (mechanism.empty()) continue; mechanisms.push_back(Decode(mechanism)); scramOffered = scramOffered || mechanism == "SCRAM-SHA-256"; } if (!scramOffered) throw DbError(std::format("unsupported SASL mechanisms {}", Dyn::ReprValue(mechanisms))); scram.emplace(url.password); const std::string first = scram->ClientFirst(); Send(Message('p', CString("SCRAM-SHA-256") + Be32(static_cast(first.size())) + first)); } else if (code == 11) { if (!scram) throw Broke("AttributeError", "'NoneType' object has no attribute 'client_final'"); Send(Message('p', scram->ClientFinal(Decode(std::string_view(body).substr(4))))); } else if (code == 12) { if (!scram) throw Broke("AttributeError", "'NoneType' object has no attribute 'verify'"); if (!scram->Verify(Decode(std::string_view(body).substr(4)))) throw DbError("the server failed SCRAM verification"); } else { throw DbError(std::format("unsupported authentication method {}", code)); } } else if (type == 'S') { const std::vector parts = Dyn::Split(body, std::string_view("\0", 1)); if (parts.size() < 2) throw Broke("ValueError", std::format("not enough values to unpack (expected 2, got {})", parts.size())); parameters[Decode(parts[0])] = Decode(parts[1]); } else if (type == 'Z') { if (body.empty()) throw Broke("IndexError", "index out of range"); status = body[0]; return; } else if (type == 'N') { notices.push_back(FieldsOf(body)); } } } Answer Collect() { Answer answer; while (true) { std::pair message; try { message = Read(); } catch (const SocketError& error) { if (error.timeout) throw DbError("no answer within the time budget"); throw DbError(std::format("the connection broke ({})", error.what())); } const auto& [type, body] = message; if (type == 'T') { const std::uint16_t count = Unpack16(body, 0); std::size_t i = 2; std::vector columns; for (std::uint16_t c = 0; c < count; ++c) { const std::size_t end = body.find('\0', i); if (end == std::string::npos) throw Broke("ValueError", "subsection not found"); columns.push_back(Dyn::DecodeReplace(std::string_view(body).substr(i, end - i))); i = end + 1 + 18; } answer.columns = std::move(columns); } else if (type == 'D') { const std::uint16_t count = Unpack16(body, 0); std::size_t i = 2; std::vector> row; for (std::uint16_t c = 0; c < count; ++c) { const auto length = static_cast(Unpack32(body, i)); i += 4; if (length < 0) { row.emplace_back(std::nullopt); } else { row.emplace_back(Dyn::DecodeReplace(std::string_view(body).substr(std::min(i, body.size()), static_cast(length)))); i += static_cast(length); } } answer.rows.push_back(std::move(row)); } else if (type == 'C') { answer.tag = Decode(body.empty() ? std::string_view() : std::string_view(body).substr(0, body.size() - 1)); } else if (type == 'I') { if (!answer.tag || answer.tag->empty()) answer.tag = "(empty)"; } else if (type == 'E') { answer.error = FieldsOf(body); } else if (type == 'N') { answer.notices.push_back(FieldsOf(body)); } else if (type == 'G') { // COPY FROM STDIN: refused before sending, kept for a server // that starts one anyway. Send(Message('f', CString("the review-gates helper sends no COPY data")) + Message('S')); } else if (type == 'Z') { if (body.empty()) throw Broke("IndexError", "index out of range"); status = body[0]; return answer; } // '1' ParseComplete, '2' BindComplete, 'n' NoData, 's' // PortalSuspended, 'S' ParameterStatus, 'A', 'K', 'H'/'d'/'c': // nothing to keep. } } Answer TakeAnswer() { try { return Collect(); } catch (const Broke& error) { throw DbError(std::format("a malformed answer ({}: {})", error.type, error.what())); } } }; Connection::Connection(std::string_view url, std::chrono::seconds timeout, std::string_view application) : state(std::make_unique()) { state->url = ParseUrl(url); const Url& p = state->url; try { state->socket = std::make_unique(p.host, static_cast(p.port), std::chrono::seconds(5)); } catch (const std::exception& error) { throw DbError(std::format("cannot connect to {}:{} ({})", p.host, p.port, ConnectErrorText(error))); } // Every failure from here on is a DbError, whatever broke: its callers // catch DbError alone. try { state->socket->SetReceiveTimeout(timeout); std::string body = Be32(196608); for (const auto& [key, value] : std::array, 4>{{ {"user", p.user}, {"database", p.database}, {"application_name", application}, {"client_encoding", "UTF8"}}}) { body += CString(key) + CString(value); } body.push_back('\0'); state->Send(Be32(static_cast(body.size() + 4)) + body); state->Authenticate(); } catch (const DbError&) { Close(); throw; } catch (const Broke& error) { Close(); throw DbError(std::format("the login exchange broke ({}: {})", error.type, error.what())); } } Connection::~Connection() { Close(); } Answer Connection::Simple(std::string_view query) { state->Send(Message('Q', CString(query))); return state->TakeAnswer(); } Answer Connection::Extended(std::string_view statement) { state->Send(Message('P', std::format("{}{}{}", std::string(1, '\0'), CString(statement), Be16(0))) + Message('B', std::format("{}{}{}{}", std::string(2, '\0'), Be16(0), Be16(0), Be16(0))) + Message('D', std::string("P\0", 2)) + Message('E', std::format("{}{}", std::string(1, '\0'), Be32(0))) + Message('S')); return state->TakeAnswer(); } void Connection::Close() { if (!state || !state->socket) return; try { const std::string terminate = Message('X'); state->socket->Send(terminate.data(), static_cast(terminate.size())); } catch (const std::exception&) { // Closing anyway. } state->socket.reset(); } char Connection::Status() const { return state->status; } const std::map& Connection::Parameters() const { return state->parameters; } Scram::Scram(std::string password, std::string user, std::string nonce) : password(std::move(password)), nonce(nonce.empty() ? Base64Encode(RandomBytes(18)) : std::move(nonce)) { bare = std::format("n={},r={}", user, this->nonce); } std::string Scram::ClientFirst() const { return std::format("n,,{}", bare); } std::string Scram::ClientFinal(std::string_view serverFirst) { const std::map a = Attributes(serverFirst); const std::string& serverNonce = Attribute(a, "r"); if (!serverNonce.starts_with(nonce)) throw DbError("SCRAM: the server nonce does not extend ours"); const std::string salt = Base64(Attribute(a, "s")); const std::string& iterationText = Attribute(a, "i"); const std::optional iterations = Dyn::ParseInt(iterationText); if (!iterations) throw Broke("ValueError", Dyn::IntError(iterationText)); if (*iterations < 1) throw Broke("ValueError", "iteration value must be greater than 0."); const std::string salted = Pbkdf2Sha256(password, salt, static_cast(*iterations)); const std::string clientKey = HmacSha256(salted, "Client Key"); const std::string withoutProof = std::format("c={},r={}", Base64Encode("n,,"), serverNonce); authMessage = std::format("{},{},{}", bare, serverFirst, withoutProof); const std::string signature = HmacSha256(Sha256(clientKey), authMessage); serverKey = HmacSha256(salted, "Server Key"); std::string proof(clientKey.size(), '\0'); for (std::size_t i = 0; i < clientKey.size(); ++i) proof[i] = static_cast(clientKey[i] ^ signature[i]); return std::format("{},p={}", withoutProof, Base64Encode(proof)); } bool Scram::Verify(std::string_view serverFinal) const { const std::map a = Attributes(serverFinal); const auto v = a.find("v"); return v != a.end() && ConstantTimeEqual(Base64(v->second), HmacSha256(serverKey, authMessage)); } std::unique_ptr Connect(std::string_view url, std::int32_t attempts, const std::function& sleep) { double delay = 0.5; std::optional last; for (std::int32_t n = 0; n < attempts; ++n) { try { return std::make_unique(url); } catch (const DbError& error) { last = error; const std::string_view text = error.what(); const bool busy = std::ranges::any_of(RetrySqlStates, [&](std::string_view state) { return text.contains(state); }); if (!busy && (n >= 2 || text.contains("login refused"))) break; // No sleep after the last attempt: nothing comes after it. if (n + 1 < attempts) { if (sleep) { sleep(delay); } else { std::this_thread::sleep_for(std::chrono::duration(delay)); } } delay = std::min(4.0, delay * 2); } } if (last) throw *last; throw DbError("no attempt made"); } // ---- output ---------------------------------------------------------------------------- std::string Cell(const std::optional& value) { if (!value) return "(null)"; const std::string v = Dyn::Replace(Dyn::Replace(Dyn::Replace(*value, "\r", "\\r"), "\n", "\\n"), "\t", "\\t"); if (Dyn::Length(v) <= static_cast(MaxCell)) return v; return std::format("{}…", Dyn::Slice(v, 0, MaxCell - 1)); } std::string Table(const std::vector& columns, const std::vector>>& rows) { std::vector> shown; for (std::size_t r = 0; r < rows.size() && r < static_cast(MaxRows); ++r) { std::vector cells; for (const std::optional& value : rows[r]) cells.push_back(Cell(value)); shown.push_back(std::move(cells)); } std::vector widths; for (std::size_t i = 0; i < columns.size(); ++i) { std::size_t width = Dyn::Length(columns[i]); for (const std::vector& row : shown) width = std::max(width, Dyn::Length(row.at(i))); widths.push_back(width); } auto line = [&](const std::vector& cells) { std::vector padded; for (std::size_t i = 0; i < cells.size(); ++i) padded.push_back(Dyn::LJust(cells[i], widths.at(i))); return Dyn::RStrip(std::format(" {}", Dyn::Join(padded, " | "))); }; std::vector out = {line(columns)}; std::vector dashes; for (const std::size_t width : widths) dashes.emplace_back(width, '-'); out.push_back(std::format("-{}-", Dyn::Join(dashes, "-+-"))); for (const std::vector& row : shown) out.push_back(line(row)); const std::size_t more = rows.size() - shown.size(); out.push_back(std::format("({} row{}{})", rows.size(), rows.size() != 1 ? "s" : "", more > 0 ? std::format(", {} not shown", more) : std::string())); return Dyn::Join(out, "\n"); } std::string ErrorText(const Fields& f, std::string_view statement) { std::vector lines = {Dyn::RStrip(std::format("{} {}: {}", FirstOf(f, 'V', 'S', "ERROR"), Get(f, 'C'), Get(f, 'M')))}; for (const auto& [key, label] : std::array, 3>{{{'D', "DETAIL"}, {'H', "HINT"}, {'W', "WHERE"}}}) { if (const std::string value = Get(f, key); !value.empty()) lines.push_back(std::format("{}: {}", label, value)); } const std::string position = Get(f, 'P'); if (!position.empty() && !statement.empty() && std::ranges::all_of(position, [](char c) { return c >= '0' && c <= '9'; })) { const std::int64_t p = Dyn::ParseInt(position).value_or(1) - 1; const std::string before = Dyn::Slice(statement, std::max(0, p - 30), p); const std::string after = Dyn::Slice(statement, p, p + 30); lines.push_back(Dyn::Replace(std::format("AT: …{}⟦{}⟧…", before, after), "\n", " ")); } return Dyn::Join(lines, "\n"); } std::string NoticeText(const Fields& f) { return std::format("{}: {}", FirstOf(f, 'V', 'S', "NOTICE"), Get(f, 'M')); } // ---- one call of the helper ------------------------------------------------------------ std::int32_t RunSql(std::string_view url, std::string_view sql, const Out& out, const Connector& connect, const Refuser& refuse) { if (const std::optional problem = CheckUrl(Json(url))) { out(std::format("{} - {}", NoDatabase, *problem)); return ExitNoDb; } if (Dyn::Length(sql) > static_cast(MaxSql)) { out(std::format("review-gates sql: the script is {} characters; the limit is {} - nothing ran", Dyn::Length(sql), MaxSql)); return ExitUsage; } const std::vector statements = SplitStatements(sql); if (statements.empty()) { out("review-gates sql: no statement given - the SQL goes on stdin, in a quoted heredoc (<<'SQL' ... SQL) - nothing ran"); return ExitUsage; } if (statements.size() > static_cast(MaxStatements)) { out(std::format("review-gates sql: {} statements; the limit is {} - nothing ran", statements.size(), MaxStatements)); return ExitUsage; } std::vector bad; for (std::size_t i = 0; i < statements.size(); ++i) { if (const std::optional why = refuse ? refuse(statements[i]) : Refused(statements[i])) bad.push_back(std::format("statement {}: {}", i + 1, *why)); } if (!bad.empty()) { out(std::format("review-gates sql: refused, nothing ran - {}", Dyn::Join(bad, "; "))); return ExitSql; } std::unique_ptr conn; try { conn = connect ? connect(url) : Connect(url); } catch (const DbError& error) { out(std::format("{} - {}", NoDatabase, error.what())); return ExitNoDb; } std::size_t size = 0; std::int32_t code = ExitOk; std::optional ended; auto say = [&](const std::string& text, bool always = false) { if (size < static_cast(MaxOutput) || always) out(text); size += Dyn::Length(text) + 1; }; // Whatever happened after the connection, the transaction is rolled // back and the call says how it ended. auto finish = [&] { try { if (conn->Status() == 'T' || conn->Status() == 'E') conn->Simple("ROLLBACK"); } catch (const std::exception&) { // Nothing more to do: the connection closes either way. } conn->Close(); out(ended ? std::format("-- the transaction ended at statement {}: what ran up to it may have been committed; the rest was not", *ended) : std::string("-- rolled back: nothing of this call was committed")); }; try { for (const std::string& query : {std::string("BEGIN"), std::format("SET LOCAL statement_timeout = '{}s'", StatementTimeoutS), std::format("SET LOCAL lock_timeout = '{}s'", LockTimeoutS), std::string("SET LOCAL idle_in_transaction_session_timeout = '60s'")}) { const Answer answer = conn->Simple(query); if (answer.error) { out(std::format("{} - the session could not be set up ({})", NoDatabase, ErrorText(*answer.error))); finish(); return ExitNoDb; } } out(std::format("{} to {} - ONE transaction, rolled back at the end of this call (statement_timeout {}s, lock_timeout {}s; at most {} rows printed per statement)", Answered, Where(url), StatementTimeoutS, LockTimeoutS, MaxRows)); for (std::size_t i = 1; i <= statements.size(); ++i) { const std::string& statement = statements[i - 1]; say(std::format("-- [{}/{}] {}", i, statements.size(), Headline(statement)), true); const Answer answer = conn->Extended(statement); for (const Fields& notice : answer.notices) say(NoticeText(notice)); if (answer.error) { say(ErrorText(*answer.error, statement), true); if (i < statements.size()) say(std::format("-- stopped: statement {} failed, so statements {}-{} did not run", i, i + 1, statements.size()), true); code = ExitSql; break; } if (answer.columns) say(Table(*answer.columns, answer.rows)); say(answer.tag && !answer.tag->empty() ? *answer.tag : std::string("(no command tag)"), true); if (conn->Status() != 'T' && conn->Status() != 'E') { ended = i; say(std::format("-- STOPPED: the transaction is no longer open after statement {} (server status {}) - nothing more runs. " "This is a pipeline defect: report it as a pipeline note; what ran up to it may have been committed.", i, conn->Status()), true); code = ExitSql; break; } } if (size > static_cast(MaxOutput)) { out(std::format("-- output cut at {} characters ({} produced): the statements ran, their rows are not all shown", MaxOutput, size)); } } catch (const DbError& error) { out(std::format("-- the database stopped answering mid-call ({}); nothing of this call was committed", error.what())); finish(); return ExitSql; } catch (...) { finish(); throw; } finish(); return code; } std::pair, std::vector>>> Query(std::string_view url, std::string_view sql) { const std::unique_ptr conn = Connect(url, 3); auto rollback = [&] { try { conn->Simple("ROLLBACK"); } catch (const std::exception&) { // Closing anyway. } conn->Close(); }; try { conn->Simple("BEGIN READ ONLY"); Answer answer = conn->Extended(sql); if (answer.error) throw SqlError(*answer.error); rollback(); return {answer.columns.value_or(std::vector()), std::move(answer.rows)}; } catch (...) { rollback(); throw; } } std::optional BypassesRls(std::string_view url) { try { const auto [columns, rows] = Query(url, "select coalesce((select rolsuper or rolbypassrls from pg_roles where rolname = current_user), false)"); return rows.at(0).at(0) == std::optional("t"); } catch (const DbError&) { return std::nullopt; } catch (const SqlError&) { return std::nullopt; } } std::pair Probe(std::string_view url) { if (const std::optional problem = CheckUrl(Json(url))) return {false, *problem}; std::vector>> rows; try { rows = Query(url, "select current_user, current_database(), version()").second; } catch (const DbError& error) { return {false, std::format("{}: {}", Redact(url), error.what())}; } catch (const SqlError& error) { return {false, std::format("{}: {}", Redact(url), error.what())}; } const std::vector>& row = rows.at(0); const std::vector version = Dyn::Split(row.at(2).value_or("")); const std::span head(version.data(), std::min(version.size(), 2)); const Url parsed = ParseUrl(url); return {true, std::format("{} on {}:{} as {} - {}", row.at(1).value_or("None"), parsed.host, parsed.port, row.at(0).value_or("None"), Dyn::Join(head, " "))}; } // ---- the broker ------------------------------------------------------------------------------ struct Broker::State { Runner run; Prober probe; std::mutex lock; std::map tokens; std::map> urls; std::unique_ptr listener; std::thread accepting; std::uint16_t port = 0; void Serve(ClientTCP& client) { bool dead = false; auto out = [&](const std::string& text) { if (dead) return; const std::string line = std::format("{}\n", Dyn::DecodeReplace(text)); try { client.Send(line.data(), static_cast(line.size())); } catch (const std::exception&) { // The helper is gone (its Bash call was stopped): the call // still runs to its ROLLBACK. dead = true; } }; // rfile.readline(limit): up to a newline, at most `limit` bytes. constexpr std::size_t Limit = 4 * static_cast(MaxSql) + 4096; std::string raw; while (raw.size() < Limit && !raw.ends_with('\n')) { std::array chunk {}; try { const std::uint32_t got = client.RecieveSync(static_cast(std::min(chunk.size(), Limit - raw.size())), chunk.data()); const std::string_view piece(chunk.data(), got); const std::size_t newline = piece.find('\n'); raw.append(newline == std::string_view::npos ? piece : piece.substr(0, newline + 1)); } catch (const std::exception&) { break; } } Json request; if (raw.ends_with('\n')) { try { request = ParseLoose(raw); } catch (const std::exception&) { request = nullptr; } } std::int32_t code = ExitOk; const Json& token = request.is_object() ? Dyn::Get(request, "token") : Dyn::EmptyDict(); std::optional url; if (request.is_object() && token.is_string()) { const std::lock_guard guard(lock); const auto found = urls.find(token.get_ref()); if (found != urls.end()) url = found->second; } if (!request.is_object()) { out(std::format("review-gates sql: the call did not arrive whole (more than {} characters, or not the helper) - nothing ran", MaxSql)); code = ExitUsage; } else if (!url) { // A token that is no string (a list, say) gets this answer too. out(std::format("{} - the driver knows no database by this token (a stale or a foreign one)", NoDatabase)); code = ExitNoDb; } else if (Dyn::Truthy(Dyn::Get(request, "probe"))) { const auto [ok, detail] = probe ? probe(*url) : Probe(*url); out(std::format("{} - {}", ok ? Answered : NoDatabase, detail)); code = ok ? ExitOk : ExitNoDb; } else { const Json& sql = Dyn::Get(request, "sql"); const std::string text = sql.is_string() ? sql.get() : std::string(); code = run ? run(*url, text, out) : RunSql(*url, text, out); } out(std::format("{}{}", ExitMark, code)); } }; Broker::Broker(Runner run, Prober probe) : state(std::make_shared()) { state->run = std::move(run); state->probe = std::move(probe); } Broker::~Broker() { Close(); } void Broker::Start() { const std::weak_ptr weak = state; state->listener = std::make_unique("127.0.0.1", 0, [weak](ClientTCP* accepted) { std::unique_ptr client(accepted); const std::shared_ptr owner = weak.lock(); if (!owner) return; // A thread per call, not Crafter.Thread's CPU-sized pool: a call // waits on the database, not on a CPU. std::thread([owner, client = std::move(client)] { try { owner->Serve(*client); } catch (const std::exception&) { // A call that broke ends here; the helper reports the // missing exit code. } }).detach(); }); state->port = state->listener->Port(); ListenerTCP* listener = state->listener.get(); state->accepting = std::thread([listener] { listener->ListenSyncSync(); }); } std::string Broker::Token(const std::string& url) { const std::lock_guard guard(state->lock); const auto found = state->tokens.find(url); if (found != state->tokens.end()) return found->second; const std::string token = TokenHex(16); state->tokens[url] = token; state->urls[token] = url; return token; } std::optional Broker::UrlOf(std::string_view token) { const std::lock_guard guard(state->lock); const auto found = state->urls.find(token); if (found == state->urls.end()) return std::nullopt; return found->second; } std::string Broker::Address(const std::string& url) { if (url.empty() || !state->listener) return ""; return std::format("127.0.0.1:{}#{}", state->port, Token(url)); } void Broker::Close() { if (!state || !state->listener) return; state->listener->Stop(); if (state->accepting.joinable()) state->accepting.join(); state->listener.reset(); } std::int32_t CallBroker(std::string_view address, std::string_view sql, bool probeOnly, const Out& out) { if (address.empty() || !address.contains('#')) { out(std::format("{} - this pass has no database (the driver gives one to a pass with the supabase environment)", NoDatabase)); return ExitNoDb; } const std::size_t hash = address.rfind('#'); const std::string_view where = address.substr(0, hash); const std::string token(address.substr(hash + 1)); const std::size_t colon = where.rfind(':'); const std::string host(colon == std::string_view::npos ? std::string_view() : where.substr(0, colon)); const std::string portText(colon == std::string_view::npos ? where : where.substr(colon + 1)); const std::optional port = Dyn::ParseInt(portText); if (!port) { out(std::format("{} - the driver's SQL broker does not answer ({})", NoDatabase, Dyn::IntError(portText))); return ExitNoDb; } // socket.create_connection raised OverflowError for it, which // call_broker did not catch: the helper ended in a traceback. if (*port < 0 || *port > 65535) throw Dyn::Error("OverflowError", "getsockaddrarg: port must be 0-65535."); std::unique_ptr socket; try { socket = std::make_unique(host, static_cast(*port), std::chrono::seconds(5)); } catch (const std::exception& error) { out(std::format("{} - the driver's SQL broker does not answer ({})", NoDatabase, ConnectErrorText(error))); return ExitNoDb; } try { // Each statement's lines come within its timeout; connecting may // retry. socket->SetReceiveTimeout(std::chrono::seconds(StatementTimeoutS + 40)); const Json request = probeOnly ? Json{{"token", token}, {"probe", true}} : Json{{"token", token}, {"sql", sql}}; const std::string line = std::format("{}\n", DumpSpaced(request)); socket->Send(line.data(), static_cast(line.size())); std::string pending; bool closed = false; while (!closed) { std::array chunk {}; try { const std::uint32_t got = socket->RecieveSync(static_cast(chunk.size()), chunk.data()); pending.append(chunk.data(), got); } catch (const SocketClosedException&) { closed = true; } std::size_t newline = 0; while ((newline = pending.find('\n')) != std::string::npos || (closed && !pending.empty())) { const std::size_t end = newline == std::string::npos ? pending.size() : newline + 1; std::string text = Dyn::RStrip(Dyn::DecodeReplace(std::string_view(pending).substr(0, end)), "\n"); pending.erase(0, end); if (text.starts_with(ExitMark)) { const std::string number = text.substr(ExitMark.size()); const std::optional code = Dyn::ParseInt(number); if (!code) throw std::invalid_argument(Dyn::IntError(number)); return static_cast(*code); } out(text); } } } catch (const SocketTimeoutException&) { out("-- the driver's SQL broker stopped answering mid-call (timed out); nothing of this call was committed"); return ExitSql; } catch (const SocketErrorException& error) { out(std::format("-- the driver's SQL broker stopped answering mid-call ({}); nothing of this call was committed", Dyn::ErrnoText(error.code))); return ExitSql; } catch (const std::exception& error) { out(std::format("-- the driver's SQL broker stopped answering mid-call ({}); nothing of this call was committed", error.what())); return ExitSql; } out("-- the driver's SQL broker closed the call without its exit code; nothing of this call was committed"); return ExitSql; } std::int32_t SqlMain(std::span args) { constexpr std::string_view Usage = R"(SQL against the database under test, for a lens with the `supabase` environment - ONE transaction per call, rolled back at its end. crafter-review sql <<'SQL' the statements on stdin (a quoted heredoc: the shell expands nothing in it) select ...; SQL crafter-review sql 'select 1' or as arguments crafter-review sql --probe does the database answer? The SQL goes to the review driver's broker, named with this pass's database in REVIEW_GATES_DB (an address and a token - the database's URL stays in the driver); the broker runs it on the local stack and sends the lines back. Exit 0 every statement ran, 1 one failed or was refused, 2 no database, 64 no statement.)"; const std::string address = Platform::GetEnv(BrokerEnv).value_or(""); const Out say = [](const std::string& line) { std::println("{}", line); std::fflush(stdout); }; if (args.size() == 1 && args[0] == "--probe") return CallBroker(address, "", true, say); if (!args.empty() && (args[0] == "-h" || args[0] == "--help")) { std::println("{}", Usage); return 0; } std::string sql; if (!args.empty()) { std::vector words(args.begin(), args.end()); sql = Dyn::Join(words, " "); } else if (!isatty(0)) { // All of stdin, as text with universal newlines. const std::string bytes((std::istreambuf_iterator(std::cin)), std::istreambuf_iterator()); sql = Dyn::UniversalNewlines(Dyn::DecodeReplace(bytes)); } return CallBroker(address, sql, false, say); } }