1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
#define _POSIX_C_SOURCE 200809L
/* Research-only server. It shares the product peer's TLS transport, wire
* format, program parser/evaluator, and store, but calls the store directly. */
#include "program.h"
#include "store.h"
#include "../../../prototypes/common/tls-link.h"
#include <openssl/err.h>
#include <openssl/sha.h>
#include <arpa/inet.h>
#include <errno.h>
#include <inttypes.h>
#include <poll.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
enum { REQUEST_BYTES = 14, REPLY_BYTES = 13, HELLO_BYTES = 40,
OP_READ = 1, OP_WRITE = 2, WANT_BODY = 0x80 };
static const unsigned char alpn[] = {10, 'p','u','r','e','-','d','o','c','/','1'};
static volatile sig_atomic_t stopping;
static void signal_stop(int signum) { (void)signum; stopping = 1; }
typedef struct { PureResult status; int64_t revision; PureSpan body; } DirectOutcome;
static uint32_t get32(const unsigned char *p) {
return ((uint32_t)p[0] << 24) | ((uint32_t)p[1] << 16) | ((uint32_t)p[2] << 8) | p[3];
}
static uint64_t get64(const unsigned char *p) {
uint64_t value = 0;
for (unsigned i = 0; i < 8; ++i) value = (value << 8) | p[i];
return value;
}
static void put32(unsigned char *p, uint32_t value) {
for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(value >> (24 - 8 * i));
}
static void put64(unsigned char *p, uint64_t value) {
for (unsigned i = 0; i < 8; ++i) p[i] = (unsigned char)(value >> (56 - 8 * i));
}
static bool number(const char *text, uint64_t limit, uint64_t *out) {
if (!text || !*text) return false;
uint64_t value = 0;
for (; *text; ++text) {
if (*text < '0' || *text > '9') return false;
unsigned digit = (unsigned)(*text - '0');
if (digit > limit || value > (limit - digit) / 10) return false;
value = value * 10 + digit;
}
*out = value;
return true;
}
static bool valid_key(const char *key) {
if (!key || !*key) return false;
size_t size = 0;
for (; size <= 64 && key[size]; ++size) {
unsigned char c = (unsigned char)key[size];
if (!((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
(c >= '0' && c <= '9') || c == '-' || c == '_')) return false;
}
return size <= 64;
}
static bool load_program(const char *path, PureProgram *program, unsigned char digest[32]) {
FILE *file = fopen(path, "rb");
if (!file) return false;
char source[PURE_PROGRAM_LIMIT + 1];
size_t length = fread(source, 1, sizeof(source), file);
bool bad = ferror(file) != 0;
if (fclose(file)) bad = true;
char error[160];
if (bad || !pure_program_parse(source, length, program, error, sizeof(error))) {
if (!bad) fprintf(stderr, "Program: %s\n", error);
return false;
}
return SHA256((const unsigned char *)source, length, digest) != NULL;
}
static bool write_hello(PureLink *link, const unsigned char digest[32], uint64_t deadline) {
unsigned char hello[HELLO_BYTES] = {'P','D','O','C',1,0,0,0};
memcpy(hello + 8, digest, 32);
return pure_link_write(link, hello, sizeof(hello), deadline);
}
static bool send_result(PureLink *link, DirectOutcome outcome, bool want_body, uint64_t deadline) {
if (outcome.status > PURE_ALLOCATION || outcome.revision < 0 ||
outcome.body.size > PURE_OBJECT_LIMIT || (outcome.body.size && !outcome.body.data)) return false;
unsigned char header[REPLY_BYTES];
header[0] = (unsigned char)outcome.status;
put64(header + 1, (uint64_t)outcome.revision);
put32(header + 9, (uint32_t)outcome.body.size);
return pure_link_write(link, header, sizeof(header), deadline) &&
(!want_body || pure_link_write(link, outcome.body.data, outcome.body.size, deadline));
}
static bool serve_request(PureLink *link, const PureProgram *program, const char *store, int *end) {
uint64_t deadline = pure_link_now_ms() + PURE_LINK_DEADLINE_MS;
unsigned char header[REQUEST_BYTES];
int got = pure_link_read(link, header, sizeof(header), deadline);
if (!got) { *end = 1; return true; }
if (got != 1) return false;
unsigned op = header[0] & ~WANT_BODY, key_size = header[1];
bool want_body = (header[0] & WANT_BODY) != 0;
uint64_t expected = get64(header + 2);
uint32_t body_size = get32(header + 10);
if ((op != OP_READ && op != OP_WRITE) || !key_size || key_size > 64 ||
expected > INT64_MAX || body_size > PURE_OBJECT_LIMIT ||
(op == OP_READ && (expected || body_size))) return false;
char key[65];
if (pure_link_read(link, key, key_size, deadline) != 1) return false;
if (memchr(key, 0, key_size)) return false;
key[key_size] = 0;
if (!valid_key(key)) {
(void)send_result(link, (DirectOutcome){.status = PURE_INVALID}, want_body, deadline);
return false;
}
DirectOutcome outcome = {0};
if (op == OP_READ) {
PureStoredView view = {0};
outcome.status = pure_store_select(store, key, program->read.offset,
program->read.length, program->read.to_end, program->read_required,
4u * 1024u * 1024u, &view);
if (outcome.status == PURE_OK) {
outcome.body = view.body;
outcome.revision = view.revision;
}
bool sent = send_result(link, outcome, want_body, deadline);
pure_store_release(&view);
return sent;
}
unsigned char *input = body_size ? malloc(body_size) : NULL;
if (body_size && !input) {
(void)send_result(link, (DirectOutcome){.status = PURE_ALLOCATION}, want_body, deadline);
return false; /* Body deliberately not consumed after rejected admission. */
}
if (body_size && pure_link_read(link, input, body_size, deadline) != 1) {
free(input);
return false;
}
PureSpan value = {0}, reply = {0}, source = {input, body_size};
size_t offset = 0, remove = 0;
if (!pure_program_input(program, source, &value, &reply) ||
(program->splice && !pure_program_splice(program, source, &offset, &remove)))
outcome.status = PURE_LIMIT;
else if (program->splice)
outcome.status = pure_store_splice(store, key, (int64_t)expected,
offset, remove, value, &outcome.revision);
else outcome.status = pure_store_put(store, key, (int64_t)expected, value,
&outcome.revision);
if (outcome.status == PURE_OK) outcome.body = reply;
bool sent = send_result(link, outcome, want_body, deadline);
free(input);
return sent;
}
static bool serve_connection(PureLink *link, const PureProgram *program, const char *store,
const unsigned char digest[32]) {
if (!write_hello(link, digest, pure_link_now_ms() + PURE_LINK_DEADLINE_MS)) return false;
int end = 0;
while (!end) if (!serve_request(link, program, store, &end)) return false;
return true;
}
static int serve(const char *bind_address, const char *port, const char *program_path,
const char *store, const char *ca, const char *cert,
const char *key, const char *client_name, unsigned max_connections) {
if (!pure_link_port_number(port, true) || !client_name || !*client_name) return 2;
PureProgram program;
unsigned char digest[32];
if (!load_program(program_path, &program, digest) || pure_store_init(store) != PURE_OK) return 2;
SSL_CTX *ctx = pure_link_context(true, ca, cert, key, alpn);
if (!ctx) { ERR_print_errors_fp(stderr); return 2; }
int listener = pure_link_listen_at(bind_address, port);
if (listener < 0) { perror("listen"); SSL_CTX_free(ctx); return 2; }
struct sockaddr_storage address;
socklen_t address_size = sizeof(address);
if (getsockname(listener, (struct sockaddr *)&address, &address_size)) abort();
unsigned actual_port = address.ss_family == AF_INET
? ntohs(((struct sockaddr_in *)&address)->sin_port)
: ntohs(((struct sockaddr_in6 *)&address)->sin6_port);
printf("READY %u\n", actual_port); fflush(stdout);
signal(SIGTERM, signal_stop); signal(SIGINT, signal_stop);
unsigned accepted = 0;
while (!stopping && (!max_connections || accepted < max_connections)) {
struct pollfd p = {.fd = listener, .events = POLLIN};
int result = poll(&p, 1, 1000);
if (result < 0) { if (errno == EINTR) continue; break; }
if (!result) continue;
int fd = accept(listener, NULL, NULL);
if (fd < 0) { if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) continue; break; }
++accepted;
PureLink link = {.fd = fd};
bool healthy = pure_link_nonblocking(fd) &&
pure_link_handshake(&link, ctx, true, client_name, alpn);
if (healthy) healthy = serve_connection(&link, &program, store, digest);
pure_link_close(&link, healthy);
}
close(listener);
SSL_CTX_free(ctx);
return 0;
}
int main(int argc, char **argv) {
signal(SIGPIPE, SIG_IGN);
if ((argc == 10 || argc == 11) && !strcmp(argv[1], "serve")) {
uint64_t limit = 0;
if (argc == 11 && (!number(argv[10], UINT32_MAX, &limit) || !limit)) return 2;
return serve(argv[2], argv[3], argv[4], argv[5], argv[6], argv[7], argv[8], argv[9],
(unsigned)limit);
}
fputs("direct-peer serve BIND_IP PORT PROGRAM STORE_DIR CA CERT KEY CLIENT_DNS [MAX_CONNECTIONS]\n", stderr);
return 2;
}
|
