Jorijn van der Graaf
53b7d6de31
Found compiling the environment-variable table for AGENTS.md: - A REVIEW_GATES_MEM_RESERVE_MB that is no number threw std::invalid_argument, which nothing caught: the process aborted where it should end with a ValueError, exit 1. nan and inf were an unchecked cast. They raise ValueError / OverflowError now; a value past int64 saturates. - An env file that is not UTF-8: `run` ended with exit 1 where its preflight should stop it (exit 3), and `seats` parsed the raw bytes. Both read it as strict UTF-8 text now. - A broker address with a port past 65535 was cut to 16 bits; it is an OverflowError now, uncaught (exit 1). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1217 lines
58 KiB
C++
1217 lines
58 KiB
C++
// SPDX-License-Identifier: AGPL-3.0-only
|
|
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
|
|
|
|
module;
|
|
#include <netdb.h>
|
|
#include <unistd.h>
|
|
#include <cstring>
|
|
#include <stdio.h>
|
|
|
|
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<std::string_view, 5> 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<std::string>& 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<const SocketTimeoutException*>(&error)) return "timed out";
|
|
if (const auto* failure = dynamic_cast<const SocketErrorException*>(&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<std::size_t>(i)] = static_cast<char>((value >> (24 - 8 * i)) & 0xFF);
|
|
return out;
|
|
}
|
|
|
|
std::string Be16(std::uint16_t value) {
|
|
return {static_cast<char>(value >> 8), static_cast<char>(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<std::uint32_t>(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<std::uint8_t>(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<std::uint16_t>((static_cast<std::uint8_t>(bytes[at]) << 8) | static_cast<std::uint8_t>(bytes[at + 1]));
|
|
}
|
|
|
|
// bytes.decode(): strict.
|
|
std::string Decode(std::string_view bytes) {
|
|
if (const std::optional<std::string> 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<std::string, std::string> Attributes(std::string_view text) {
|
|
std::map<std::string, std::string> 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<std::string, std::string>& 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<std::string_view, 2> 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<std::int32_t> port = split.Port();
|
|
url.port = port && *port != 0 ? *port : 5432;
|
|
const std::optional<std::string> 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<std::string> CheckUrl(const Json& url) {
|
|
if (!url.is_string() || url.get_ref<const std::string&>().empty()) return "no database URL";
|
|
Url parsed;
|
|
try {
|
|
parsed = ParseUrl(url.get_ref<const std::string&>());
|
|
} 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<std::string> 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<std::string> 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<std::u32string_view> 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<std::string> 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<std::int64_t>(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<std::string> 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<std::size_t> 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<std::string> FirstWords(std::string_view text, std::size_t k) {
|
|
const std::u32string statement = Dyn::Decode(text);
|
|
const std::u32string_view s(statement);
|
|
std::vector<std::string> 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<char>(ch - 'A' + 'a');
|
|
}
|
|
words.push_back(std::move(word));
|
|
i = j;
|
|
}
|
|
return words;
|
|
}
|
|
|
|
std::optional<std::string> Refused(std::string_view statement) {
|
|
const std::vector<std::string> w = FirstWords(statement, 4);
|
|
const std::string k = w.empty() ? std::string() : w[0];
|
|
std::span<const std::string> rest = w.empty() ? std::span<const std::string>() : std::span<const std::string>(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<std::int64_t>(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<ClientTCP> socket;
|
|
std::string buffer;
|
|
std::size_t position = 0;
|
|
char status = '\0';
|
|
std::vector<Fields> notices;
|
|
std::map<std::string, std::string> 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<char, 65536> chunk {};
|
|
try {
|
|
const std::uint32_t got = socket->RecieveSync(static_cast<std::uint32_t>(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<char, std::string> 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<std::uint32_t>(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> scram;
|
|
while (true) {
|
|
std::pair<char, std::string> 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<std::uint32_t>(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<std::string> 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<char, std::string> 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<std::string> 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<std::optional<std::string>> row;
|
|
for (std::uint16_t c = 0; c < count; ++c) {
|
|
const auto length = static_cast<std::int32_t>(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<std::size_t>(length))));
|
|
i += static_cast<std::size_t>(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>()) {
|
|
state->url = ParseUrl(url);
|
|
const Url& p = state->url;
|
|
try {
|
|
state->socket = std::make_unique<ClientTCP>(p.host, static_cast<std::uint16_t>(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<std::pair<std::string_view, std::string_view>, 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<std::uint32_t>(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<std::uint32_t>(terminate.size()));
|
|
} catch (const std::exception&) {
|
|
// Closing anyway.
|
|
}
|
|
state->socket.reset();
|
|
}
|
|
|
|
char Connection::Status() const {
|
|
return state->status;
|
|
}
|
|
|
|
const std::map<std::string, std::string>& 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<std::string, std::string> 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<std::int64_t> 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<std::uint32_t>(*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<char>(clientKey[i] ^ signature[i]);
|
|
return std::format("{},p={}", withoutProof, Base64Encode(proof));
|
|
}
|
|
|
|
bool Scram::Verify(std::string_view serverFinal) const {
|
|
const std::map<std::string, std::string> a = Attributes(serverFinal);
|
|
const auto v = a.find("v");
|
|
return v != a.end() && ConstantTimeEqual(Base64(v->second), HmacSha256(serverKey, authMessage));
|
|
}
|
|
|
|
std::unique_ptr<Connection> Connect(std::string_view url, std::int32_t attempts, const std::function<void(double)>& sleep) {
|
|
double delay = 0.5;
|
|
std::optional<DbError> last;
|
|
for (std::int32_t n = 0; n < attempts; ++n) {
|
|
try {
|
|
return std::make_unique<Connection>(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<double>(delay));
|
|
}
|
|
}
|
|
delay = std::min(4.0, delay * 2);
|
|
}
|
|
}
|
|
if (last) throw *last;
|
|
throw DbError("no attempt made");
|
|
}
|
|
|
|
// ---- output ----------------------------------------------------------------------------
|
|
std::string Cell(const std::optional<std::string>& 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<std::size_t>(MaxCell)) return v;
|
|
return std::format("{}…", Dyn::Slice(v, 0, MaxCell - 1));
|
|
}
|
|
|
|
std::string Table(const std::vector<std::string>& columns, const std::vector<std::vector<std::optional<std::string>>>& rows) {
|
|
std::vector<std::vector<std::string>> shown;
|
|
for (std::size_t r = 0; r < rows.size() && r < static_cast<std::size_t>(MaxRows); ++r) {
|
|
std::vector<std::string> cells;
|
|
for (const std::optional<std::string>& value : rows[r]) cells.push_back(Cell(value));
|
|
shown.push_back(std::move(cells));
|
|
}
|
|
std::vector<std::size_t> widths;
|
|
for (std::size_t i = 0; i < columns.size(); ++i) {
|
|
std::size_t width = Dyn::Length(columns[i]);
|
|
for (const std::vector<std::string>& row : shown) width = std::max(width, Dyn::Length(row.at(i)));
|
|
widths.push_back(width);
|
|
}
|
|
auto line = [&](const std::vector<std::string>& cells) {
|
|
std::vector<std::string> 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<std::string> out = {line(columns)};
|
|
std::vector<std::string> dashes;
|
|
for (const std::size_t width : widths) dashes.emplace_back(width, '-');
|
|
out.push_back(std::format("-{}-", Dyn::Join(dashes, "-+-")));
|
|
for (const std::vector<std::string>& 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<std::string> lines = {Dyn::RStrip(std::format("{} {}: {}", FirstOf(f, 'V', 'S', "ERROR"), Get(f, 'C'), Get(f, 'M')))};
|
|
for (const auto& [key, label] : std::array<std::pair<char, std::string_view>, 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<std::int64_t>(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<std::string> problem = CheckUrl(Json(url))) {
|
|
out(std::format("{} - {}", NoDatabase, *problem));
|
|
return ExitNoDb;
|
|
}
|
|
if (Dyn::Length(sql) > static_cast<std::size_t>(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<std::string> 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<std::size_t>(MaxStatements)) {
|
|
out(std::format("review-gates sql: {} statements; the limit is {} - nothing ran", statements.size(), MaxStatements));
|
|
return ExitUsage;
|
|
}
|
|
std::vector<std::string> bad;
|
|
for (std::size_t i = 0; i < statements.size(); ++i) {
|
|
if (const std::optional<std::string> 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<Connection> 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<std::size_t> ended;
|
|
auto say = [&](const std::string& text, bool always = false) {
|
|
if (size < static_cast<std::size_t>(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<std::size_t>(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<std::string>, std::vector<std::vector<std::optional<std::string>>>> Query(std::string_view url, std::string_view sql) {
|
|
const std::unique_ptr<Connection> 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::string>()), std::move(answer.rows)};
|
|
} catch (...) {
|
|
rollback();
|
|
throw;
|
|
}
|
|
}
|
|
|
|
std::optional<bool> 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<std::string>("t");
|
|
} catch (const DbError&) {
|
|
return std::nullopt;
|
|
} catch (const SqlError&) {
|
|
return std::nullopt;
|
|
}
|
|
}
|
|
|
|
std::pair<bool, std::string> Probe(std::string_view url) {
|
|
if (const std::optional<std::string> problem = CheckUrl(Json(url))) return {false, *problem};
|
|
std::vector<std::vector<std::optional<std::string>>> 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<std::optional<std::string>>& row = rows.at(0);
|
|
const std::vector<std::string> version = Dyn::Split(row.at(2).value_or(""));
|
|
const std::span<const std::string> head(version.data(), std::min<std::size_t>(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<std::string, std::string> tokens;
|
|
std::map<std::string, std::string, std::less<>> urls;
|
|
std::unique_ptr<ListenerTCP> 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<std::uint32_t>(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<std::size_t>(MaxSql) + 4096;
|
|
std::string raw;
|
|
while (raw.size() < Limit && !raw.ends_with('\n')) {
|
|
std::array<char, 65536> chunk {};
|
|
try {
|
|
const std::uint32_t got = client.RecieveSync(static_cast<std::uint32_t>(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<std::string> url;
|
|
if (request.is_object() && token.is_string()) {
|
|
const std::lock_guard guard(lock);
|
|
const auto found = urls.find(token.get_ref<const std::string&>());
|
|
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>() : 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>()) {
|
|
state->run = std::move(run);
|
|
state->probe = std::move(probe);
|
|
}
|
|
|
|
Broker::~Broker() {
|
|
Close();
|
|
}
|
|
|
|
void Broker::Start() {
|
|
const std::weak_ptr<State> weak = state;
|
|
state->listener = std::make_unique<ListenerTCP>("127.0.0.1", 0, [weak](ClientTCP* accepted) {
|
|
std::unique_ptr<ClientTCP> client(accepted);
|
|
const std::shared_ptr<State> 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<std::string> 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<std::int64_t> 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<ClientTCP> socket;
|
|
try {
|
|
socket = std::make_unique<ClientTCP>(host, static_cast<std::uint16_t>(*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<std::uint32_t>(line.size()));
|
|
std::string pending;
|
|
bool closed = false;
|
|
while (!closed) {
|
|
std::array<char, 65536> chunk {};
|
|
try {
|
|
const std::uint32_t got = socket->RecieveSync(static_cast<std::uint32_t>(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<std::int64_t> code = Dyn::ParseInt(number);
|
|
if (!code) throw std::invalid_argument(Dyn::IntError(number));
|
|
return static_cast<std::int32_t>(*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<const std::string_view> 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<std::string> 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<char>(std::cin)), std::istreambuf_iterator<char>());
|
|
sql = Dyn::UniversalNewlines(Dyn::DecodeReplace(bytes));
|
|
}
|
|
return CallBroker(address, sql, false, say);
|
|
}
|
|
}
|