Jorijn van der Graaf
51bdd6298a
Crafter.Network master now carries, in their final shape, the pieces the HTTP layer and the SQL broker were written against on a side branch: ClientHTTP1::Fetch follows redirects by RFC 9110 §15.4 on a URL type of its own, and SocketErrorException is a std::system_error with the phase that failed. The broker and the HTTP layer read the errno or the resolver code from code() and the phase from the field, instead of matching message prefixes, and the pin moves to that commit, so the build no longer needs --local against a checkout of the side branch. The messages the run log, the preflight detail and the SQL helper show are unchanged, but for the wording of a refused URL.
1218 lines
58 KiB
C++
1218 lines
58 KiB
C++
// SPDX-License-Identifier: AGPL-3.0-only
|
|
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
|
|
|
|
module;
|
|
#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)) {
|
|
// "[Errno N] text": the resolver's own text for a resolver
|
|
// code, the C library's for an errno.
|
|
if (failure->code().category() == ResolverCategory()) return std::format("[Errno {}] {}", failure->code().value(), failure->code().message());
|
|
return Dyn::ErrnoText(failure->code().value());
|
|
}
|
|
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().value()), 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().value())));
|
|
} 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().value())));
|
|
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);
|
|
}
|
|
}
|