|
| 1 | +/* |
| 2 | + * Licensed to the Apache Software Foundation (ASF) under one |
| 3 | + * or more contributor license agreements. See the NOTICE file |
| 4 | + * distributed with this work for additional information |
| 5 | + * regarding copyright ownership. The ASF licenses this file |
| 6 | + * to you under the Apache License, Version 2.0 (the |
| 7 | + * "License"); you may not use this file except in compliance |
| 8 | + * with the License. You may obtain a copy of the License at |
| 9 | + * |
| 10 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | + * |
| 12 | + * Unless required by applicable law or agreed to in writing, |
| 13 | + * software distributed under the License is distributed on an |
| 14 | + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | + * KIND, either express or implied. See the License for the |
| 16 | + * specific language governing permissions and limitations |
| 17 | + * under the License. |
| 18 | + */ |
| 19 | + |
| 20 | +#include "iceberg/catalog/rest/http_client.h" |
| 21 | + |
| 22 | +#include <array> |
| 23 | +#include <atomic> |
| 24 | +#include <cctype> |
| 25 | +#include <cstddef> |
| 26 | +#include <format> |
| 27 | +#include <string> |
| 28 | +#include <string_view> |
| 29 | +#include <thread> |
| 30 | +#include <utility> |
| 31 | + |
| 32 | +#include <gmock/gmock.h> |
| 33 | +#include <gtest/gtest.h> |
| 34 | + |
| 35 | +#include "iceberg/catalog/rest/auth/auth_session.h" |
| 36 | +#include "iceberg/catalog/rest/error_handlers.h" |
| 37 | +#include "iceberg/test/matchers.h" |
| 38 | + |
| 39 | +#ifndef _WIN32 |
| 40 | +# include <poll.h> |
| 41 | +# include <unistd.h> |
| 42 | + |
| 43 | +# include <arpa/inet.h> |
| 44 | +# include <netinet/in.h> |
| 45 | +# include <sys/socket.h> |
| 46 | +# include <sys/time.h> |
| 47 | +#endif |
| 48 | + |
| 49 | +namespace iceberg::rest { |
| 50 | + |
| 51 | +namespace { |
| 52 | + |
| 53 | +#ifndef _WIN32 |
| 54 | + |
| 55 | +/// \brief Minimal HTTP/1.1 server on 127.0.0.1 that answers every request with |
| 56 | +/// a fixed response and closes the connection. |
| 57 | +class LoopbackHttpServer { |
| 58 | + public: |
| 59 | + explicit LoopbackHttpServer(std::string response) : response_(std::move(response)) {} |
| 60 | + |
| 61 | + ~LoopbackHttpServer() { |
| 62 | + stop_ = true; |
| 63 | + if (thread_.joinable()) { |
| 64 | + thread_.join(); |
| 65 | + } |
| 66 | + if (listen_fd_ >= 0) { |
| 67 | + close(listen_fd_); |
| 68 | + } |
| 69 | + } |
| 70 | + |
| 71 | + LoopbackHttpServer(const LoopbackHttpServer&) = delete; |
| 72 | + LoopbackHttpServer& operator=(const LoopbackHttpServer&) = delete; |
| 73 | + |
| 74 | + /// \brief Bind an ephemeral port and start serving. Returns false on failure. |
| 75 | + bool Start() { |
| 76 | + listen_fd_ = socket(AF_INET, SOCK_STREAM, 0); |
| 77 | + if (listen_fd_ < 0) { |
| 78 | + return false; |
| 79 | + } |
| 80 | + int reuse = 1; |
| 81 | + setsockopt(listen_fd_, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); |
| 82 | + |
| 83 | + sockaddr_in addr{}; |
| 84 | + addr.sin_family = AF_INET; |
| 85 | + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); |
| 86 | + addr.sin_port = 0; |
| 87 | + if (bind(listen_fd_, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) != 0 || |
| 88 | + listen(listen_fd_, 8) != 0) { |
| 89 | + return false; |
| 90 | + } |
| 91 | + socklen_t len = sizeof(addr); |
| 92 | + if (getsockname(listen_fd_, reinterpret_cast<sockaddr*>(&addr), &len) != 0) { |
| 93 | + return false; |
| 94 | + } |
| 95 | + port_ = ntohs(addr.sin_port); |
| 96 | + thread_ = std::thread([this] { Serve(); }); |
| 97 | + return true; |
| 98 | + } |
| 99 | + |
| 100 | + int port() const { return port_; } |
| 101 | + int request_count() const { return request_count_.load(); } |
| 102 | + std::string url(std::string_view path) const { |
| 103 | + return std::format("http://127.0.0.1:{}{}", port_, path); |
| 104 | + } |
| 105 | + |
| 106 | + private: |
| 107 | + void Serve() { |
| 108 | + while (!stop_) { |
| 109 | + pollfd pfd{.fd = listen_fd_, .events = POLLIN, .revents = 0}; |
| 110 | + if (poll(&pfd, 1, /*timeout_ms=*/50) <= 0) { |
| 111 | + continue; |
| 112 | + } |
| 113 | + int conn = accept(listen_fd_, nullptr, nullptr); |
| 114 | + if (conn < 0) { |
| 115 | + continue; |
| 116 | + } |
| 117 | + ReadRequest(conn); |
| 118 | + ++request_count_; |
| 119 | + SendAll(conn, response_); |
| 120 | + close(conn); |
| 121 | + } |
| 122 | + } |
| 123 | + |
| 124 | + /// \brief Consume the request headers and any Content-Length body so the |
| 125 | + /// client does not see a connection reset before it reads the response. |
| 126 | + static void ReadRequest(int conn) { |
| 127 | + timeval timeout{.tv_sec = 2, .tv_usec = 0}; |
| 128 | + setsockopt(conn, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); |
| 129 | + |
| 130 | + std::string request; |
| 131 | + std::array<char, 4096> buf{}; |
| 132 | + size_t header_end = std::string::npos; |
| 133 | + size_t content_length = 0; |
| 134 | + while (true) { |
| 135 | + ssize_t n = recv(conn, buf.data(), buf.size(), 0); |
| 136 | + if (n <= 0) { |
| 137 | + return; |
| 138 | + } |
| 139 | + request.append(buf.data(), static_cast<size_t>(n)); |
| 140 | + if (header_end == std::string::npos) { |
| 141 | + header_end = request.find("\r\n\r\n"); |
| 142 | + if (header_end == std::string::npos) { |
| 143 | + continue; |
| 144 | + } |
| 145 | + content_length = ParseContentLength(request.substr(0, header_end)); |
| 146 | + } |
| 147 | + if (request.size() >= header_end + 4 + content_length) { |
| 148 | + return; |
| 149 | + } |
| 150 | + } |
| 151 | + } |
| 152 | + |
| 153 | + static size_t ParseContentLength(std::string headers) { |
| 154 | + for (auto& c : headers) { |
| 155 | + c = static_cast<char>(std::tolower(static_cast<unsigned char>(c))); |
| 156 | + } |
| 157 | + constexpr std::string_view kHeader = "content-length:"; |
| 158 | + size_t pos = headers.find(kHeader); |
| 159 | + if (pos == std::string::npos) { |
| 160 | + return 0; |
| 161 | + } |
| 162 | + return std::stoul(headers.substr(pos + kHeader.size())); |
| 163 | + } |
| 164 | + |
| 165 | + static void SendAll(int conn, const std::string& data) { |
| 166 | +# ifdef MSG_NOSIGNAL |
| 167 | + constexpr int kFlags = MSG_NOSIGNAL; |
| 168 | +# else |
| 169 | + constexpr int kFlags = 0; |
| 170 | + int no_sigpipe = 1; |
| 171 | + setsockopt(conn, SOL_SOCKET, SO_NOSIGPIPE, &no_sigpipe, sizeof(no_sigpipe)); |
| 172 | +# endif |
| 173 | + size_t sent = 0; |
| 174 | + while (sent < data.size()) { |
| 175 | + ssize_t n = send(conn, data.data() + sent, data.size() - sent, kFlags); |
| 176 | + if (n <= 0) { |
| 177 | + return; |
| 178 | + } |
| 179 | + sent += static_cast<size_t>(n); |
| 180 | + } |
| 181 | + } |
| 182 | + |
| 183 | + std::string response_; |
| 184 | + int listen_fd_ = -1; |
| 185 | + int port_ = 0; |
| 186 | + std::atomic<bool> stop_{false}; |
| 187 | + std::atomic<int> request_count_{0}; |
| 188 | + std::thread thread_; |
| 189 | +}; |
| 190 | + |
| 191 | +#endif // _WIN32 |
| 192 | + |
| 193 | +} // namespace |
| 194 | + |
| 195 | +// A 3xx response must be surfaced to the caller as an error and must not be |
| 196 | +// followed: following it would replay the request, including all headers, to |
| 197 | +// a location the caller did not choose. |
| 198 | +TEST(HttpClientTest, DoesNotFollowRedirects) { |
| 199 | +#ifdef _WIN32 |
| 200 | + GTEST_SKIP() << "Loopback test server is only implemented for POSIX"; |
| 201 | +#else |
| 202 | + LoopbackHttpServer target( |
| 203 | + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n" |
| 204 | + "Content-Length: 2\r\nConnection: close\r\n\r\n{}"); |
| 205 | + ASSERT_TRUE(target.Start()); |
| 206 | + |
| 207 | + LoopbackHttpServer redirector( |
| 208 | + std::format("HTTP/1.1 302 Found\r\nLocation: {}\r\nContent-Length: 0\r\n" |
| 209 | + "Connection: close\r\n\r\n", |
| 210 | + target.url("/v1/config"))); |
| 211 | + ASSERT_TRUE(redirector.Start()); |
| 212 | + |
| 213 | + HttpClient client; |
| 214 | + auto session = auth::AuthSession::MakeDefault({{"Authorization", "Bearer token"}}); |
| 215 | + const auto& handler = *DefaultErrorHandler::Instance(); |
| 216 | + const std::string url = redirector.url("/v1/config"); |
| 217 | + |
| 218 | + auto expect_not_followed = [&](const Result<HttpResponse>& result, |
| 219 | + std::string_view method) { |
| 220 | + SCOPED_TRACE(method); |
| 221 | + EXPECT_THAT(result, IsError(ErrorKind::kRestError)); |
| 222 | + EXPECT_THAT(result, HasErrorMessage("code: 302")); |
| 223 | + EXPECT_EQ(target.request_count(), 0); |
| 224 | + }; |
| 225 | + |
| 226 | + expect_not_followed(client.Get(url, {}, {}, handler, *session), "GET"); |
| 227 | + expect_not_followed(client.Post(url, "{}", {}, handler, *session), "POST"); |
| 228 | + expect_not_followed( |
| 229 | + client.PostForm(url, {{"grant_type", "client_credentials"}}, {}, handler, *session), |
| 230 | + "POST form"); |
| 231 | + expect_not_followed(client.Head(url, {}, handler, *session), "HEAD"); |
| 232 | + expect_not_followed(client.Delete(url, {}, {}, handler, *session), "DELETE"); |
| 233 | + |
| 234 | + EXPECT_EQ(redirector.request_count(), 5); |
| 235 | +#endif |
| 236 | +} |
| 237 | + |
| 238 | +} // namespace iceberg::rest |
0 commit comments