/root/doris/be/src/service/http/utils.cpp
Line | Count | Source |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | |
18 | | #include "service/http/utils.h" |
19 | | |
20 | | #include <absl/strings/str_split.h> |
21 | | #include <fcntl.h> |
22 | | #include <stdint.h> |
23 | | #include <sys/stat.h> |
24 | | #include <unistd.h> |
25 | | |
26 | | #include <ostream> |
27 | | #include <string> |
28 | | #include <unordered_map> |
29 | | #include <vector> |
30 | | |
31 | | #include "common/config.h" |
32 | | #include "common/logging.h" |
33 | | #include "common/status.h" |
34 | | #include "common/utils.h" |
35 | | #include "io/fs/file_system.h" |
36 | | #include "io/fs/local_file_system.h" |
37 | | #include "load/group_commit/wal/wal_manager.h" |
38 | | #include "runtime/exec_env.h" |
39 | | #include "service/http/http_channel.h" |
40 | | #include "service/http/http_client.h" |
41 | | #include "service/http/http_common.h" |
42 | | #include "service/http/http_headers.h" |
43 | | #include "service/http/http_method.h" |
44 | | #include "service/http/http_request.h" |
45 | | #include "service/http/http_status.h" |
46 | | #include "util/md5.h" |
47 | | #include "util/path_util.h" |
48 | | #include "util/security.h" |
49 | | #include "util/string_util.h" |
50 | | #include "util/url_coding.h" |
51 | | |
52 | | namespace doris { |
53 | | |
54 | | const uint32_t CHECK_SUPPORT_TIMEOUT = 3; |
55 | | const uint32_t DOWNLOAD_FILE_MAX_RETRY = 3; |
56 | | const uint32_t LIST_REMOTE_FILE_TIMEOUT = 15; |
57 | | |
58 | 3 | std::string encode_basic_auth(const std::string& user, const std::string& passwd) { |
59 | 3 | std::string auth = user + ":" + passwd; |
60 | 3 | std::string encoded_auth; |
61 | 3 | base64_encode(auth, &encoded_auth); |
62 | 3 | static std::string s_prefix = "Basic "; |
63 | 3 | return s_prefix + encoded_auth; |
64 | 3 | } |
65 | | |
66 | 52 | bool parse_basic_auth(const HttpRequest& req, std::string* user, std::string* passwd) { |
67 | | // const auto& token = req.header(HttpHeaders::AUTH_TOKEN); |
68 | | |
69 | 52 | const char k_basic[] = "Basic "; |
70 | 52 | const auto& auth = req.header(HttpHeaders::AUTHORIZATION); |
71 | 52 | if (auth.compare(0, sizeof(k_basic) - 1, k_basic, sizeof(k_basic) - 1) != 0) { |
72 | 20 | return false; |
73 | 20 | } |
74 | 32 | std::string encoded_str = auth.substr(sizeof(k_basic) - 1); |
75 | 32 | std::string decoded_auth; |
76 | 32 | if (!base64_decode(encoded_str, &decoded_auth)) { |
77 | 1 | return false; |
78 | 1 | } |
79 | 31 | auto pos = decoded_auth.find(':'); |
80 | 31 | if (pos == std::string::npos) { |
81 | 1 | return false; |
82 | 1 | } |
83 | 30 | user->assign(decoded_auth.c_str(), pos); |
84 | 30 | passwd->assign(decoded_auth.c_str() + pos + 1); |
85 | | |
86 | 30 | return true; |
87 | 31 | } |
88 | | |
89 | 41 | bool parse_basic_auth(const HttpRequest& req, AuthInfo* auth) { |
90 | | // deprecated, removed in 3.1, use AUTH_TOKEN |
91 | 41 | const auto& token = req.header("token"); |
92 | | // deprecated, removed in 3.1, use AUTH_TOKEN |
93 | 41 | const auto& auth_code = req.header(HTTP_AUTH_CODE); |
94 | 41 | const auto& auth_token = req.header(HttpHeaders::AUTH_TOKEN); |
95 | | |
96 | 41 | std::tuple<std::string, std::string, std::string> tmp; |
97 | 41 | auto& [user, pass, cluster] = tmp; |
98 | 41 | bool valid_basic_auth = parse_basic_auth(req, &user, &pass); |
99 | 41 | if (valid_basic_auth) { // always set the basic auth, the user may be useful |
100 | 23 | auto pos = user.find('@'); |
101 | 23 | if (pos != std::string::npos) { |
102 | 0 | cluster.assign(user.c_str() + pos + 1); |
103 | 0 | user.assign(user.c_str(), pos); // user is updated |
104 | 0 | } |
105 | 23 | auth->user = user; |
106 | 23 | auth->passwd = pass; |
107 | 23 | auth->cluster = cluster; |
108 | 23 | } |
109 | | |
110 | 41 | if (!token.empty()) { |
111 | 0 | auth->token = token; // deprecated |
112 | 41 | } else if (!auth_token.empty()) { |
113 | 3 | auth->token = auth_token; |
114 | 38 | } else if (!auth_code.empty()) { |
115 | | // auth_code comes straight from a request header, a malformed one is just an |
116 | | // invalid credential and must not throw out of the http callback. The value is a |
117 | | // credential that is_sensitive_header() masks elsewhere, so it is not logged. |
118 | 2 | auto parsed_auth_code = safe_stoll(auth_code, HTTP_AUTH_CODE); |
119 | 2 | if (!parsed_auth_code.has_value()) { |
120 | 1 | LOG(WARNING) << "parse auth code failed, malformed " << HTTP_AUTH_CODE << " header"; |
121 | 1 | return false; |
122 | 1 | } |
123 | 1 | auth->auth_code = parsed_auth_code.value(); // deprecated |
124 | 36 | } else if (!valid_basic_auth) { |
125 | 15 | return false; |
126 | 15 | } |
127 | | |
128 | | // set user ip |
129 | 25 | auth->user_ip.assign(req.remote_host() != nullptr ? req.remote_host() : ""); |
130 | | |
131 | 25 | auth->cert_pem = req.header(HTTP_HEADER_CLIENT_CERT_PEM); |
132 | 25 | auth->cert_subject = req.header(HTTP_HEADER_CLIENT_CERT_SUBJECT); |
133 | 25 | auth->cert_san = req.header(HTTP_HEADER_CLIENT_CERT_SAN); |
134 | 25 | auth->cert_issuer = req.header(HTTP_HEADER_CLIENT_CERT_ISSUER); |
135 | 25 | auth->cert_cipher = req.header(HTTP_HEADER_CLIENT_CERT_CIPHER); |
136 | 25 | auth->cert_validity_not_before = req.header(HTTP_HEADER_CLIENT_CERT_NOT_BEFORE); |
137 | 25 | auth->cert_validity_not_after = req.header(HTTP_HEADER_CLIENT_CERT_NOT_AFTER); |
138 | | |
139 | 25 | return true; |
140 | 41 | } |
141 | | |
142 | | // Do a simple decision, only deal a few type |
143 | 4 | std::string get_content_type(const std::string& file_name) { |
144 | 4 | std::string file_ext = path_util::file_extension(file_name); |
145 | 4 | VLOG_TRACE << "file_name: " << file_name << "; file extension: [" << file_ext << "]"; |
146 | 4 | if (file_ext == std::string(".html") || file_ext == std::string(".htm")) { |
147 | 0 | return "text/html; charset=utf-8"; |
148 | 4 | } else if (file_ext == std::string(".js")) { |
149 | 0 | return "application/javascript; charset=utf-8"; |
150 | 4 | } else if (file_ext == std::string(".css")) { |
151 | 0 | return "text/css; charset=utf-8"; |
152 | 4 | } else if (file_ext == std::string(".txt")) { |
153 | 0 | return "text/plain; charset=utf-8"; |
154 | 4 | } else if (file_ext == std::string(".png")) { |
155 | 0 | return "image/png"; |
156 | 4 | } else if (file_ext == std::string(".ico")) { |
157 | 0 | return "image/x-icon"; |
158 | 4 | } else { |
159 | 4 | return "text/plain; charset=utf-8"; |
160 | 4 | } |
161 | 4 | } |
162 | | |
163 | | void do_file_response(const std::string& file_path, HttpRequest* req, |
164 | 4 | bufferevent_rate_limit_group* rate_limit_group, bool is_acquire_md5) { |
165 | 4 | if (file_path.find("..") != std::string::npos) { |
166 | 0 | LOG(WARNING) << "Not allowed to read relative path: " << file_path; |
167 | 0 | HttpChannel::send_error(req, HttpStatus::FORBIDDEN); |
168 | 0 | return; |
169 | 0 | } |
170 | | |
171 | | // read file content and send response |
172 | 4 | int fd = open(file_path.c_str(), O_RDONLY); |
173 | 4 | if (fd < 0) { |
174 | 0 | LOG(WARNING) << "Failed to open file: " << file_path; |
175 | 0 | HttpChannel::send_error(req, HttpStatus::NOT_FOUND); |
176 | 0 | return; |
177 | 0 | } |
178 | 4 | struct stat st; |
179 | 4 | auto res = fstat(fd, &st); |
180 | 4 | if (res < 0) { |
181 | 0 | close(fd); |
182 | 0 | LOG(WARNING) << "Failed to open file: " << file_path; |
183 | 0 | HttpChannel::send_error(req, HttpStatus::NOT_FOUND); |
184 | 0 | return; |
185 | 0 | } |
186 | | |
187 | 4 | int64_t file_size = st.st_size; |
188 | | |
189 | | // TODO(lingbin): process "IF_MODIFIED_SINCE" header |
190 | | // TODO(lingbin): process "RANGE" header |
191 | 4 | const std::string& range_header = req->header(HttpHeaders::RANGE); |
192 | 4 | if (!range_header.empty()) { |
193 | | // analyse range header |
194 | 0 | } |
195 | | |
196 | 4 | req->add_output_header(HttpHeaders::CONTENT_TYPE, get_content_type(file_path).c_str()); |
197 | | |
198 | 4 | if (is_acquire_md5) { |
199 | 1 | Md5Digest md5; |
200 | | |
201 | 1 | void* buf = mmap(nullptr, file_size, PROT_READ, MAP_SHARED, fd, 0); |
202 | 1 | md5.update(buf, file_size); |
203 | 1 | md5.digest(); |
204 | 1 | munmap(buf, file_size); |
205 | | |
206 | 1 | req->add_output_header(HttpHeaders::CONTENT_MD5, md5.hex().c_str()); |
207 | 1 | } |
208 | | |
209 | 4 | if (req->method() == HttpMethod::HEAD) { |
210 | 3 | close(fd); |
211 | 3 | req->add_output_header(HttpHeaders::CONTENT_LENGTH, std::to_string(file_size).c_str()); |
212 | 3 | HttpChannel::send_reply(req); |
213 | 3 | return; |
214 | 3 | } |
215 | | |
216 | 1 | HttpChannel::send_file(req, fd, 0, file_size, rate_limit_group); |
217 | 1 | } |
218 | | |
219 | 2 | void do_dir_response(const std::string& dir_path, HttpRequest* req, bool is_acquire_filesize) { |
220 | 2 | bool exists = true; |
221 | 2 | std::vector<io::FileInfo> files; |
222 | 2 | Status st = io::global_local_filesystem()->list(dir_path, true, &files, &exists); |
223 | 2 | if (!st.ok()) { |
224 | 0 | LOG(WARNING) << "Failed to scan dir. " << st; |
225 | 0 | HttpChannel::send_error(req, HttpStatus::INTERNAL_SERVER_ERROR); |
226 | 0 | return; |
227 | 0 | } |
228 | | |
229 | 2 | VLOG_DEBUG << "list dir: " << dir_path << ", exists: " << exists |
230 | 0 | << ", file count: " << files.size(); |
231 | | |
232 | 2 | if (!exists) { |
233 | 0 | HttpChannel::send_error(req, HttpStatus::NOT_FOUND); |
234 | 0 | return; |
235 | 0 | } |
236 | | |
237 | 2 | const std::string FILE_DELIMITER_IN_DIR_RESPONSE = "\n"; |
238 | | |
239 | 2 | std::stringstream result; |
240 | 37 | for (auto& file : files) { |
241 | 37 | result << file.file_name << FILE_DELIMITER_IN_DIR_RESPONSE; |
242 | 37 | if (is_acquire_filesize) { |
243 | 35 | result << file.file_size << FILE_DELIMITER_IN_DIR_RESPONSE; |
244 | 35 | } |
245 | 37 | } |
246 | | |
247 | 2 | std::string result_str = result.str(); |
248 | 2 | HttpChannel::send_reply(req, result_str); |
249 | 2 | } |
250 | | |
251 | 2 | bool load_size_smaller_than_wal_limit(int64_t content_length) { |
252 | | // 1. req->header(HttpHeaders::CONTENT_LENGTH) will return streamload content length. If it is empty or equals to 0, it means this streamload |
253 | | // is a chunked streamload and we are not sure its size. |
254 | | // 2. if streamload content length is too large, like larger than 80% of the WAL constrain. |
255 | | // |
256 | | // This two cases, we are not certain that the Write-Ahead Logging (WAL) constraints allow for writing down |
257 | | // these blocks within the limited space. So we need to set group_commit = false to avoid dead lock. |
258 | 2 | size_t max_available_size = ExecEnv::GetInstance()->wal_mgr()->get_max_available_size(); |
259 | 2 | return (content_length < 0.8 * max_available_size); |
260 | 2 | } |
261 | | |
262 | 1 | Status is_support_batch_download(const std::string& endpoint) { |
263 | 1 | std::string url = fmt::format("http://{}/api/_tablet/_batch_download?check=true", endpoint); |
264 | 1 | auto check_support_cb = [&url](HttpClient* client) { |
265 | 1 | RETURN_IF_ERROR(client->init(url)); |
266 | 1 | client->set_timeout_ms(CHECK_SUPPORT_TIMEOUT * 1000); |
267 | 1 | client->set_method(HttpMethod::HEAD); |
268 | 1 | std::string response; |
269 | 1 | return client->execute(&response); |
270 | 1 | }; |
271 | 1 | return HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1, check_support_cb); |
272 | 1 | } |
273 | | |
274 | | Status list_remote_files_v2(const std::string& address, const std::string& token, |
275 | | const std::string& remote_dir, |
276 | 1 | std::vector<std::pair<std::string, size_t>>* file_info_list) { |
277 | 1 | std::string remote_url = |
278 | 1 | fmt::format("http://{}/api/_tablet/_batch_download?token={}&dir={}&list=true", address, |
279 | 1 | token, remote_dir); |
280 | | |
281 | 1 | std::string file_list_str; |
282 | 1 | auto list_files_cb = [&](HttpClient* client) { |
283 | 1 | file_list_str.clear(); |
284 | 1 | RETURN_IF_ERROR(client->init(remote_url, false)); |
285 | 1 | client->set_method(HttpMethod::GET); |
286 | 1 | client->set_timeout_ms(LIST_REMOTE_FILE_TIMEOUT * 1000); |
287 | 1 | return client->execute(&file_list_str); |
288 | 1 | }; |
289 | 1 | Status status = HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1, list_files_cb); |
290 | 1 | if (!status.ok()) { |
291 | 0 | LOG(WARNING) << "failed to list remote files from " << remote_url |
292 | 0 | << ", status: " << status.to_string() << ", response: " << file_list_str; |
293 | 0 | return status; |
294 | 0 | } |
295 | | |
296 | 1 | std::vector<std::string> file_list = |
297 | 1 | absl::StrSplit(file_list_str, "\n", absl::SkipWhitespace()); |
298 | 1 | if (file_list.size() % 2 != 0) { |
299 | 0 | return Status::InternalError("batch download files: invalid file list, size is not even"); |
300 | 0 | } |
301 | | |
302 | 1 | VLOG_DEBUG << "list remote files from " << remote_url |
303 | 0 | << ", file count: " << file_list.size() / 2; |
304 | | |
305 | 36 | for (size_t i = 0; i < file_list.size(); i += 2) { |
306 | 35 | uint64_t file_size = 0; |
307 | 35 | try { |
308 | 35 | file_size = std::stoull(file_list[i + 1]); |
309 | 35 | } catch (std::exception&) { |
310 | 0 | return Status::InternalError("batch download files: invalid file size format: " + |
311 | 0 | file_list[i + 1]); |
312 | 0 | } |
313 | 35 | file_info_list->emplace_back(std::move(file_list[i]), file_size); |
314 | 35 | } |
315 | | |
316 | 1 | return Status::OK(); |
317 | 1 | } |
318 | | |
319 | | Status download_files_v2(const std::string& address, const std::string& token, |
320 | | const std::string& remote_dir, const std::string& local_dir, |
321 | 1 | const std::vector<std::pair<std::string, size_t>>& file_info_list) { |
322 | 1 | std::string remote_url = fmt::format("http://{}/api/_tablet/_batch_download?dir={}&token={}", |
323 | 1 | address, remote_dir, token); |
324 | | |
325 | 1 | size_t batch_file_size = 0; |
326 | 1 | std::unordered_set<std::string> expected_files; |
327 | 1 | std::stringstream ss; |
328 | 35 | for (const auto& file_info : file_info_list) { |
329 | 35 | ss << file_info.first << "\n"; |
330 | 35 | batch_file_size += file_info.second; |
331 | 35 | expected_files.insert(file_info.first); |
332 | 35 | } |
333 | 1 | std::string payload = ss.str(); |
334 | | |
335 | 1 | uint64_t estimate_timeout = batch_file_size / config::download_low_speed_limit_kbps / 1024; |
336 | 1 | if (estimate_timeout < config::download_low_speed_time) { |
337 | 1 | estimate_timeout = config::download_low_speed_time; |
338 | 1 | } |
339 | | |
340 | 1 | LOG(INFO) << "begin to download files from " << remote_url << " to " << local_dir |
341 | 1 | << ", file count: " << file_info_list.size() << ", total size: " << batch_file_size |
342 | 1 | << ", timeout: " << estimate_timeout; |
343 | | |
344 | 1 | auto callback = [&](HttpClient* client) -> Status { |
345 | 1 | RETURN_IF_ERROR(client->init(remote_url, false)); |
346 | 1 | client->set_method(HttpMethod::POST); |
347 | 1 | client->set_payload(payload); |
348 | 1 | client->set_timeout_ms(estimate_timeout * 1000); |
349 | 1 | RETURN_IF_ERROR(client->download_multi_files(local_dir, expected_files)); |
350 | 35 | for (auto&& [file_name, file_size] : file_info_list) { |
351 | 35 | std::string local_file_path = local_dir + "/" + file_name; |
352 | | |
353 | 35 | std::error_code ec; |
354 | | // Check file length |
355 | 35 | uint64_t local_file_size = std::filesystem::file_size(local_file_path, ec); |
356 | 35 | if (ec) { |
357 | 0 | LOG(WARNING) << "download file error: " << ec.message(); |
358 | 0 | return Status::IOError("can't retrive file_size of {}, due to {}", local_file_path, |
359 | 0 | ec.message()); |
360 | 0 | } |
361 | 35 | if (local_file_size != file_size) { |
362 | 0 | LOG(WARNING) << "download file length error" |
363 | 0 | << ", remote_path=" << mask_token(remote_url) |
364 | 0 | << ", file_name=" << file_name << ", file_size=" << file_size |
365 | 0 | << ", local_file_size=" << local_file_size; |
366 | 0 | return Status::InternalError("downloaded file size is not equal"); |
367 | 0 | } |
368 | 35 | RETURN_IF_ERROR(io::global_local_filesystem()->permission( |
369 | 35 | local_file_path, io::LocalFileSystem::PERMS_OWNER_RW)); |
370 | 35 | } |
371 | | |
372 | 1 | return Status::OK(); |
373 | 1 | }; |
374 | 1 | return HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1, callback); |
375 | 1 | } |
376 | | |
377 | | } // namespace doris |