diff options
| -rw-r--r-- | CHECKPOINT.md | 2 | ||||
| -rw-r--r-- | docs/design/008-module-format.md | 3 | ||||
| -rw-r--r-- | docs/research/083-delivery-over-http.md | 9 | ||||
| -rw-r--r-- | prototypes/connections/Makefile | 2 | ||||
| -rw-r--r-- | prototypes/connections/deliver.c | 113 | ||||
| -rw-r--r-- | prototypes/connections/deliver.h | 7 | ||||
| -rw-r--r-- | prototypes/connections/serve.c | 4 |
7 files changed, 107 insertions, 33 deletions
diff --git a/CHECKPOINT.md b/CHECKPOINT.md index b50858d..e7cb1c0 100644 --- a/CHECKPOINT.md +++ b/CHECKPOINT.md @@ -18,7 +18,7 @@ | HTTP 서비스 | 보기가 GET, 입력이 POST다(JSON 본문과 그 안의 그룹 객체를 제자리에서 읽는다, 라우터 코드 없음). 앱의 `send`는 이벤트와 함께 저장되고, `--send 출력=URL`로 다른 앱이나 HTTP 끝점에 순서대로 묶어 전해지며, 받는 앱은 번호마다 한 번 적용한다(서버를 죽였다 살려도 정확히 한 번). 답은 저장 뒤에 나간다. 상태를 가진 앱은 소유자 스레드 하나가 입력을 적용하고, 연결 스레드는 마지막 저장의 스냅숏으로 보기를 답한다(쓰기가 동기화하는 동안에도 읽기는 기다리지 않는다) | 008 HTTP 호스트, [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [082](docs/research/082-json-in-place.md), [083](docs/research/083-delivery-over-http.md) | | 성능(Pi 5) | 문서 서버: 작은 GET 151.6k/s, 1 MiB GET 2,794/s(axum 1,372, drogon 708), 충돌 안전 PUT은 axum-atomic의 5.1–5.8배. 앱 서비스(aup 22줄 대 손으로 쓴 서비스): 내구성 있는 쓰기 약 76배(그 뒤 저장소 커밋을 한 구간으로 바꿔 64연결 21–26k/s, axum·drogon 내구 약 180/s), 20행 JSON GET 145k/s(axum 95k, drogon 144k), 쓰기와 섞인 읽기 149k/s·p99 0.6 ms(axum·drogon 내구 1.1–1.3k/s·p99 111–156 ms), 최대 RSS 3.0–4.1 MiB(axum 6.4, drogon 11.6) | [075](docs/research/075-http-host-against-axum.md), [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [081](docs/research/081-tail-records.md) | -남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 전달의 TLS와 응답을 이벤트로 돌려받기(083 다음 결정), 앱 라우트의 인증·TLS(문서 +남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 응답을 이벤트로 돌려받기(083 다음 결정), 앱 라우트의 인증·TLS(문서 피어의 TLS는 있다), aup·Vega의 배포 묶음. ## 공개할 작업물과 완료 증거 diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md index 07cff23..96a77fa 100644 --- a/docs/design/008-module-format.md +++ b/docs/design/008-module-format.md @@ -338,7 +338,8 @@ text·리스트가 남을 때만 위치를 유지한다. 이벤트 30 → 50, p50은 오히려 5.4 → 3.5 ms). 끊김은 답을 돌려준 뒤에 온 요청이 하나라도 있어야 세기 시작한다. 저장을 기다리는 동안 줄에 선 요청만으로 끊김을 세면 막 답을 받은 클라이언트가 다음 묶음을 놓치고, 클라이언트가 두 무리로 갈려 번갈아 저장한다(연결 스레드 둘, 64연결에서 묶음당 34개, 15.8–16.0k/s였고, 고친 뒤 61.6개, 22.9–23.9k/s). -- **보냄의 전달.** `--send 출력=http://호스트:포트/경로`로 출력을 받는 곳에 잇는다. 저장이 끝날 때마다 소유자가 새 보냄을 +- **보냄의 전달.** `--send 출력=http[s]://호스트:포트/경로`로 출력을 받는 곳에 잇는다(`https`는 시스템 CA나 `--send-ca 파일`로 + 인증서와 호스트 이름을 검증한다). 저장이 끝날 때마다 소유자가 새 보냄을 JSON(필드와 그룹)으로 그 출력의 줄에 넣고, 출력마다 전달 스레드가 순서대로 보낸다(`X-Pure-Source: 이름.출력`, `X-Pure-Sequence`, `Idempotency-Key`). 2xx 또는 Pure 받는 쪽의 거절이면 전해진 것이고, 소유자가 덜어 낸다. 그 밖에는 쉬었다가 다시 보낸다(버리지 않는다). 받는 쪽이 `X-Pure-Batch`를 알리면 밀린 것을 묶어 보낸다([083](../research/083-delivery-over-http.md)). diff --git a/docs/research/083-delivery-over-http.md b/docs/research/083-delivery-over-http.md index 48da056..5427590 100644 --- a/docs/research/083-delivery-over-http.md +++ b/docs/research/083-delivery-over-http.md @@ -37,7 +37,9 @@ 받는 쪽은 한 요청의 메시지를 차례로 한 묶음에서 적용하고 한 번 저장한다. 거절한 것은 들은 것으로 치고 넘어가고, 다른 실패에서 멈추면 묶음 전체를 다시 받는다(앞부분은 번호로 걸러진다). 제3자 끝점은 이 헤더를 모르므로 하나씩 받는다. - JSON 본문은 출력의 필드와 그룹 그대로다([082](082-json-in-place.md)의 입력과 같은 모양). 한 그룹의 필드가 떨어져 있는 - 출력, `https`는 시작할 때 거절한다. UTF-8이 아닌 bytes 필드는 JSON으로 만들 수 없어 그 출력이 거기서 멈춘다(보냄은 남는다). + 출력은 시작할 때 거절한다. UTF-8이 아닌 bytes 필드는 JSON으로 만들 수 없어 그 출력이 거기서 멈춘다(보냄은 남는다). +- **TLS.** `https://` 받는 곳에는 OpenSSL로 TLS 1.2 이상을 쓴다. 인증서는 시스템 CA(또는 `--send-ca 파일`)로 검증하고, 호스트 + 이름(SNI와 `SSL_set1_host`, IP 주소면 그 주소)과 맞아야 한다. 맞지 않으면 기록하고 10초마다 다시 시도한다(보냄은 남는다). ## 장치가 말한 것 @@ -58,13 +60,14 @@ Raspberry Pi 5. 받는 서버는 CPU 0–1(연결 스레드 2), 보내는 서버 `kill -9`로 죽이고 같은 저장소로 다시 띄운다. 그 뒤 수량 150(받는 앱이 거절한다)과 7의 주문을 더 넣는다. 받는 앱의 건수와 합계가 보내는 앱이 저장한 것(거절한 한 건을 빼고)과 같아지고, 확인 응답한 주문은 모두 그 안에 있다. 다섯 번(각 약 7,300건)과 ASan·UBSan 빌드로 한 번 통과했다. 거절한 주문 뒤의 주문도 전해졌다(막히지 않는다). +- aup `make check`의 `server_webhook`: 파이썬 HTTPS 서버(자체 인증서)를 제3자 웹훅으로 두고, 주문 다섯 건이 순서대로 + 출력의 JSON 모양과 헤더(`X-Pure-Source`, `Idempotency-Key`, 묶음 헤더 없음)로 도착한다. 처음에 503으로 답한 2번은 다시 온다. + 그 인증서를 믿지 않는 보내는 쪽에서는 아무것도 도착하지 않고 기록이 남는다. ASan·UBSan 빌드로도 통과했다. - `test-module`의 링크 검사([078](078-app-connections.md), 008)가 런타임 쪽(이벤트의 보냄, 저장 실패, 재시작, 중복·거절)을 맡는다. ## 다음 결정 -- **TLS.** 제3자 API는 거의 `https`다. 문서 피어의 TLS([055](055-authenticated-peer-transport-cost.md))를 전달 클라이언트에 - 쓰는 것이 다음이다. - **응답을 이벤트로.** 지금은 보내고 끝이다(알림, 웹훅). 결제처럼 답이 필요한 호출은 응답을 앱의 입력으로 돌려주는 배선(`--send 출력=URL --reply 입력`)이 필요하다. - 저장이 끝날 때마다 outbox 전체를 훑어 새 보냄을 찾는다. 받는 쪽이 오래 꺼져 수만 건이 밀리면 저장마다 그만큼 훑는다. diff --git a/prototypes/connections/Makefile b/prototypes/connections/Makefile index 60d83d7..fe575b5 100644 --- a/prototypes/connections/Makefile +++ b/prototypes/connections/Makefile @@ -45,7 +45,7 @@ $(BUILD)/pure-connections-async-gated: server.c $(HOST) flow.c program.c $(CORE) # The HTTP host for aup modules: persistent connections, epoll, group commits. $(BUILD)/pure-serve: serve.c deliver.c flow.c module-store.c $(MODULE) $(CORE) deliver.h module-store.h $(MODULE_HEADERS) | $(BUILD) - $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_ASYNC_EFFECTS serve.c deliver.c flow.c module-store.c $(MODULE) $(CORE) $(LDFLAGS) -lm $(LDLIBS) -o $@ + $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_ASYNC_EFFECTS serve.c deliver.c flow.c module-store.c $(MODULE) $(CORE) $(LDFLAGS) -lssl -lcrypto -lm $(LDLIBS) -o $@ $(BUILD)/pure-connections-fixed: server.c $(HOST) flow.c $(CORE) $(HEADERS) $(BUILD)/embedded-plan.h | $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_EMBEDDED_PROGRAM -I$(BUILD) server.c $(HOST) flow.c $(CORE) $(LDFLAGS) $(LDLIBS) -o $@ diff --git a/prototypes/connections/deliver.c b/prototypes/connections/deliver.c index 5f32978..c4136a2 100644 --- a/prototypes/connections/deliver.c +++ b/prototypes/connections/deliver.c @@ -5,6 +5,9 @@ #include <netdb.h> #include <netinet/in.h> #include <netinet/tcp.h> +#include <openssl/ssl.h> +#include <openssl/x509v3.h> +#include <arpa/inet.h> #include <pthread.h> #include <stdio.h> #include <stdlib.h> @@ -16,7 +19,7 @@ #include <unistd.h> enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256 }; -enum { DELIVERED, AGAIN, REFUSED_BY_PEER }; +enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED }; typedef struct { char *body; size_t size; uint64_t seq; } Pending; @@ -34,9 +37,32 @@ struct PureWire { pthread_t thread; void (*notify)(void *); void *context; - int fd; /* the connection kept for the next send, or -1; changed under the lock */ + int fd; /* the connection kept for the next send, or -1; changed under the lock */ + bool tls; /* https: the connection is TLS */ + SSL *ssl; /* its TLS session */ }; +/* ---- TLS: one client context, verifying the receiver's certificate against + * the system's CAs or the file pure_wire_trust named, and its name. */ +static pthread_once_t tls_once = PTHREAD_ONCE_INIT; +static SSL_CTX *tls_context; +static const char *tls_ca; + +void pure_wire_trust(const char *ca_file) { tls_ca = ca_file; } + +static void make_tls_context(void) { + SSL_CTX *context = SSL_CTX_new(TLS_client_method()); + if (!context) return; + if (!SSL_CTX_set_min_proto_version(context, TLS1_2_VERSION) || + !(tls_ca ? SSL_CTX_load_verify_locations(context, tls_ca, NULL) : SSL_CTX_set_default_verify_paths(context))) { + SSL_CTX_free(context); + return; + } + SSL_CTX_set_verify(context, SSL_VERIFY_PEER, NULL); + SSL_CTX_set_mode(context, SSL_MODE_AUTO_RETRY); + tls_context = context; +} + /* ---- A send as JSON: its fields in order, a group's fields as an object. */ typedef struct { char *bytes; size_t size, capacity; bool failed; } Text; @@ -169,15 +195,22 @@ PureWire *pure_wire_open(const PureModule *m, const char *output, const char *ur snprintf(error, size, "a source name has 1 to %d bytes", PURE_SOURCE); return NULL; } - if (strncmp(url, "http://", 7)) { - snprintf(error, size, strncmp(url, "https://", 8) ? "a URL is http://host[:port]/path" : - "https is not delivered to yet: use http"); + bool tls = !strncmp(url, "https://", 8); + if (!tls && strncmp(url, "http://", 7)) { + snprintf(error, size, "a URL is http[s]://host[:port]/path"); return NULL; } - const char *host = url + 7, *slash = strchr(host, '/'), *colon = memchr(host, ':', slash ? (size_t)(slash - host) : strlen(host)); + if (tls) { + pthread_once(&tls_once, make_tls_context); + if (!tls_context) { + snprintf(error, size, "no TLS context: %s cannot be read", tls_ca ? tls_ca : "the system's CAs"); + return NULL; + } + } + const char *host = url + (tls ? 8 : 7), *slash = strchr(host, '/'), *colon = memchr(host, ':', slash ? (size_t)(slash - host) : strlen(host)); const char *host_end = colon ? colon : slash ? slash : host + strlen(host); - const char *port = colon ? colon + 1 : "80"; - size_t port_size = colon ? (size_t)((slash ? slash : host + strlen(host)) - port) : 2; + const char *port = colon ? colon + 1 : tls ? "443" : "80"; + size_t port_size = colon ? (size_t)((slash ? slash : host + strlen(host)) - port) : strlen(port); if (host_end == host || (size_t)(host_end - host) >= 256 || !port_size || port_size >= 8 || (slash && strlen(slash) >= 1024)) { snprintf(error, size, "a URL is http://host[:port]/path"); @@ -192,6 +225,7 @@ PureWire *pure_wire_open(const PureModule *m, const char *output, const char *ur snprintf(w->path, sizeof(w->path), "%s", slash ? slash : "/"); memcpy(w->source, source, length + 1); w->fd = -1; + w->tls = tls; pthread_mutex_init(&w->lock, NULL); pthread_cond_init(&w->changed, NULL); return w; @@ -250,13 +284,15 @@ bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) { /* ---- HTTP/1.1, one request at a time on a kept connection. */ -typedef struct { int fd; char bytes[4096]; size_t at, end; } Reader; +typedef struct { PureWire *wire; char bytes[4096]; size_t at, end; } Reader; static bool fill(Reader *r) { if (r->at == r->end) r->at = r->end = 0; if (r->end == sizeof(r->bytes)) return false; + PureWire *w = r->wire; ssize_t n; - do n = recv(r->fd, r->bytes + r->end, sizeof(r->bytes) - r->end, 0); while (n < 0 && errno == EINTR); + if (w->ssl) n = SSL_read(w->ssl, r->bytes + r->end, (int)(sizeof(r->bytes) - r->end)); + else do n = recv(w->fd, r->bytes + r->end, sizeof(r->bytes) - r->end, 0); while (n < 0 && errno == EINTR); if (n <= 0) return false; r->end += (size_t)n; return true; @@ -294,10 +330,11 @@ static bool skip(Reader *r, uint64_t size) { return true; } -static bool send_all(int fd, const char *bytes, size_t size) { +static bool send_all(PureWire *w, const char *bytes, size_t size) { while (size) { - ssize_t n = send(fd, bytes, size, MSG_NOSIGNAL); - if (n < 0 && errno == EINTR) continue; + ssize_t n = w->ssl ? SSL_write(w->ssl, bytes, (int)(size < INT32_MAX ? size : INT32_MAX)) : + send(w->fd, bytes, size, MSG_NOSIGNAL); + if (n < 0 && !w->ssl && errno == EINTR) continue; if (n <= 0) return false; bytes += n; size -= (size_t)n; @@ -305,6 +342,22 @@ static bool send_all(int fd, const char *bytes, size_t size) { return true; } +/* TLS over a connected socket: the receiver's certificate must name the host + * (a DNS name, or an IP address written as one). */ +static SSL *start_tls(const PureWire *w, int fd) { + SSL *ssl = SSL_new(tls_context); + if (!ssl) return NULL; + unsigned char address[16]; + bool literal = inet_pton(AF_INET, w->host, address) == 1 || inet_pton(AF_INET6, w->host, address) == 1; + bool named = literal ? X509_VERIFY_PARAM_set1_ip_asc(SSL_get0_param(ssl), w->host) == 1 : + SSL_set_tlsext_host_name(ssl, w->host) == 1 && SSL_set1_host(ssl, w->host) == 1; + if (!named || SSL_set_fd(ssl, fd) != 1 || SSL_connect(ssl) != 1 || SSL_get_verify_result(ssl) != X509_V_OK) { + SSL_free(ssl); + return NULL; + } + return ssl; +} + static int connect_to(const PureWire *w) { struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_STREAM}, *list; if (getaddrinfo(w->host, w->port, &hints, &list)) return -1; @@ -328,7 +381,9 @@ static int connect_to(const PureWire *w) { static void drop_connection(PureWire *w) { pthread_mutex_lock(&w->lock); + if (w->ssl) SSL_free(w->ssl); if (w->fd >= 0) close(w->fd); + w->ssl = NULL; w->fd = -1; pthread_mutex_unlock(&w->lock); } @@ -342,8 +397,8 @@ static bool refusal(const char *error) { /* The answer to one request: its status, whether a Pure receiver refused it, * and whether the connection may carry the next. */ -static bool read_answer(int fd, int *status, bool *refused, bool *keep, unsigned *batch) { - Reader r = {.fd = fd}; +static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, unsigned *batch) { + Reader r = {.wire = w}; char line[1024]; if (!read_line(&r, line, sizeof(line)) || strncmp(line, "HTTP/1.", 7) || strlen(line) < 12) return false; *status = atoi(line + 9); @@ -406,15 +461,23 @@ static int post(PureWire *w, const Pending *p, unsigned count) { bool stopping = w->stopping; pthread_mutex_unlock(&w->lock); if (stopping) return AGAIN; + SSL *ssl = w->tls ? start_tls(w, fd) : NULL; + if (w->tls && !ssl) { + drop_connection(w); + return TLS_FAILED; + } + pthread_mutex_lock(&w->lock); + w->ssl = ssl; + pthread_mutex_unlock(&w->lock); } int status; - bool refused, keep, sent = send_all(w->fd, head, (size_t)size); + bool refused, keep, sent = send_all(w, head, (size_t)size); unsigned batch_size = 0; for (unsigned i = 0; sent && i < count; ++i) - sent = (!i || count == 1 || send_all(w->fd, ",", 1)) && (i || count == 1 || send_all(w->fd, "[", 1)) && - send_all(w->fd, p[i].body, p[i].size); - if (sent && count > 1) sent = send_all(w->fd, "]", 1); - if (!sent || !read_answer(w->fd, &status, &refused, &keep, &batch_size)) { + sent = (!i || count == 1 || send_all(w, ",", 1)) && (i || count == 1 || send_all(w, "[", 1)) && + send_all(w, p[i].body, p[i].size); + if (sent && count > 1) sent = send_all(w, "]", 1); + if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size)) { drop_connection(w); if (fresh) return AGAIN; continue; @@ -458,12 +521,13 @@ static void *deliver(void *context) { pthread_mutex_lock(&w->lock); continue; } - if (outcome == REFUSED_BY_PEER && reported != next[0].seq) { - fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"the receiver does not take it; trying again\"}\n", - w->source, (unsigned long long)next[0].seq); + if ((outcome == REFUSED_BY_PEER || outcome == TLS_FAILED) && reported != next[0].seq) { + fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"%s\"}\n", w->source, + (unsigned long long)next[0].seq, outcome == TLS_FAILED ? "the receiver's certificate is not trusted; trying again" : + "the receiver does not take it; trying again"); reported = next[0].seq; } - pause_ms = outcome == REFUSED_BY_PEER ? LONGEST_PAUSE_MS : pause_ms ? 2 * pause_ms : FIRST_PAUSE_MS; + pause_ms = outcome != AGAIN ? LONGEST_PAUSE_MS : pause_ms ? 2 * pause_ms : FIRST_PAUSE_MS; if (pause_ms > LONGEST_PAUSE_MS) pause_ms = LONGEST_PAUSE_MS; struct timespec until; clock_gettime(CLOCK_REALTIME, &until); @@ -491,6 +555,7 @@ void pure_wire_close(PureWire *w) { pthread_cond_broadcast(&w->changed); pthread_mutex_unlock(&w->lock); if (w->started) pthread_join(w->thread, NULL); + if (w->ssl) SSL_free(w->ssl); if (w->fd >= 0) close(w->fd); for (size_t i = 0; i < w->count; ++i) free(w->queue[w->head + i].body); free(w->queue); diff --git a/prototypes/connections/deliver.h b/prototypes/connections/deliver.h index f720296..13048fc 100644 --- a/prototypes/connections/deliver.h +++ b/prototypes/connections/deliver.h @@ -15,8 +15,11 @@ * once, so a send delivered twice (its answer lost) is harmless. */ typedef struct PureWire PureWire; -/* A wire for the output named output of m to url (http://host[:port]/path), - * as source; NULL with why in error when it cannot be. */ +/* A wire for the output named output of m to url (http[s]://host[:port]/path), + * as source; NULL with why in error when it cannot be. An https receiver's + * certificate must be trusted (the system's CAs, or the file given to + * pure_wire_trust before the first https wire) and name its host. */ +void pure_wire_trust(const char *ca_file); PureWire *pure_wire_open(const PureModule *m, const char *output, const char *url, const char *source, char *error, size_t size); uint32_t pure_wire_output(const PureWire *); diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c index 2ecb780..d175e29 100644 --- a/prototypes/connections/serve.c +++ b/prototypes/connections/serve.c @@ -1839,6 +1839,7 @@ int main(int argc, char **argv) { else if (!strcmp(argv[i], "--store")) store = argv[i + 1]; else if (!strcmp(argv[i], "--send") && send_count < PURE_MODULE_INPUTS) sends[send_count++] = argv[i + 1]; else if (!strcmp(argv[i], "--name")) name = argv[i + 1]; + else if (!strcmp(argv[i], "--send-ca")) pure_wire_trust(argv[i + 1]); else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value; else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value; else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value; @@ -1948,6 +1949,7 @@ int main(int argc, char **argv) { return 0; usage: fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] " - "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=http://host:port/path ...] [--name NAME]\n", stderr); + "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=http[s]://host:port/path ...] [--name NAME] " + "[--send-ca FILE]\n", stderr); return 2; } |