?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
proxy.c: declare functions and file-scoped variables static
[iwhd.git] / rest.c
blobf5016749691c02d98d07aeb7a055a996d02f31fc
1 /* Copyright (C) 2010 Red Hat, Inc.
2
3 This program is free software: you can redistribute it and/or modify
4 it under the terms of the GNU General Public License as published by
5 the Free Software Foundation, either version 3 of the License, or
6 (at your option) any later version.
7
8 This program is distributed in the hope that it will be useful,
9 but WITHOUT ANY WARRANTY; without even the implied warranty of
10 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11 GNU General Public License for more details.
12
13 You should have received a copy of the GNU General Public License
14 along with this program. If not, see <http://www.gnu.org/licenses/>. */
15
16 #include <config.h>
17
18 #include <error.h>
19 #include <fcntl.h>
20 #include <getopt.h>
21 #include <poll.h>
22 #include <pthread.h>
23 #include <semaphore.h>
24 #include <stdint.h>
25 #include <stdio.h>
26 #include <stdlib.h>
27 #include <string.h>
28 #include <strings.h>
29 #include <unistd.h>
30 #include <sys/stat.h>
31 #include <assert.h>
32
33 #include <microhttpd.h>
34 #include <hstor.h> /* only for ARRAY_SIZE at this point */
35 #include <curl/curl.h>
36 #include <glib.h>
37
38 #include "iwh.h"
39 #include "meta.h"
40 #include "backend.h"
41 #include "proxy.h"
42 #include "template.h"
43 #include "mpipe.h"
44 #include "state_defs.h"
45
46 /* Define-away for now. Eventually, define to gettext. */
47 #define _(msgid) (msgid)
48
49 #if defined(DEBUG)
50 #define MY_MHD_FLAGS MHD_USE_THREAD_PER_CONNECTION | MHD_USE_DEBUG
51 //#define MY_MHD_FLAGS MHD_USE_SELECT_INTERNALLY | MHD_USE_DEBUG
52 #else
53 #define MY_MHD_FLAGS MHD_USE_THREAD_PER_CONNECTION
54 #endif
55
56 extern backend_func_tbl bad_func_tbl;
57 extern backend_func_tbl s3_func_tbl;
58 extern backend_func_tbl curl_func_tbl;
59 extern backend_func_tbl fs_func_tbl;
60
61 typedef enum {
62 URL_ROOT=0, URL_BUCKET, URL_OBJECT, URL_ATTR, URL_INVAL,
63 URL_QUERY, URL_PROVLIST
64 } url_type;
65
66 typedef struct {
67 char *method;
68 url_type utype;
69 MHD_AccessHandlerCallback handler;
70 } rule;
71
72 static int fs_mode = 0;
73 static unsigned short my_port = MY_PORT;
74 const char *program_name;
75
76 static char *(reserved_name[]) = { "_default", "_query", "_new", NULL };
77 static char *(reserved_attr[]) = { "bucket", "key", "date", "etag", "loc", NULL };
78
79 static backend_func_tbl *main_func_tbl = &bad_func_tbl;
80
81 static void
82 free_ms (my_state *ms)
83 {
84 if (ms->cleanup & CLEANUP_CURL) {
85 curl_easy_cleanup(ms->curl);
86 }
87
88 if (ms->cleanup & CLEANUP_BUF_PTR) {
89 free(ms->pipe.data_ptr);
90 }
91
92 if (ms->cleanup & CLEANUP_POST) {
93 MHD_destroy_post_processor(ms->post);
94 }
95
96 if (ms->cleanup & CLEANUP_DICT) {
97 g_hash_table_destroy(ms->dict);
98 }
99
100 if (ms->cleanup & CLEANUP_QUERY) {
101 meta_query_stop(ms->query);
102 }
103
104 if (ms->cleanup & CLEANUP_TMPL) {
105 free(ms->gen_ctx);
106 }
107
108 if (ms->cleanup & CLEANUP_URL) {
109 free(ms->url);
110 }
111
112 free(ms);
113 }
114
115 static int
116 validate_put (struct MHD_Connection *conn)
117 {
118 const char *mhdr;
119
120 if (!master_host) {
121 /* We're not a slave, so we don't care. */
122 return 1;
123 }
124
125 mhdr = MHD_lookup_connection_value(conn,MHD_HEADER_KIND,
126 "X-redhat-role");
127 return (mhdr && !strcmp(mhdr,"master"));
128 }
129
130 static int
131 is_reserved (char *cand, char **resv_list)
132 {
133 int i;
134
135 for (i = 0; resv_list[i]; ++i) {
136 if (!strcmp(cand,resv_list[i])) {
137 return TRUE;
138 }
139 }
140
141 return FALSE;
142 }
143
144 static int
145 validate_url (const char *url)
146 {
147 char *slash = rindex(url,'/');
148
149 if (!slash) {
150 /* There should be at least one betwixt bucket and key. */
151 return 0;
152 }
153
154 return !is_reserved(slash+1,reserved_name);
155 }
156
157 /**********
158 * The proxy has MHD on one side and CURL on the other. The CURL side is
159 * always run in a child thread. Yes, there are both context switches
160 * and copies between the threads. Get over it. The focus here is on
161 * supporting multi-way replication on PUT, with minimal complexity. These
162 * were the two easiest libraries to use, and they both want to allocate
163 * their own buffers so we're kind of stuck with the copies unless we want
164 * to buffer whole files in memory (which would have to be virtual since
165 * they're potentialy bigger than physical) or explicitly ping them through
166 * a local filesystem. We could potentially take over scheduling from one
167 * or both to avoid some of the context switching, but those interfaces are
168 * significantly more error-prone and (at least in CURL's case) buggy.
169 *
170 * For a GET, the CURL child acts as a producer while the MHD parent acts
171 * as consumer. For a PUT, the MHD parent is the producer and the CURL
172 * child is the consumer. For GET the MHD component is invoked via a
173 * callback set up in the access handler; for PUT it's invoked via repeated
174 * calls to the access handler itself. Either way, the producer's job is
175 * to post its pointer+length to the my_state structure and then wait for
176 * all consumers to check back in saying they've finished it. This might
177 * involve multiple passes through each consumer for one pass through the
178 * single producer. When the producer is done, it does a similar handshake
179 * with the consumers. Each consumer has its own pipe_private structure,
180 * containing a pointer to the shared my_state plus a per-consumer offset
181 * into the current chunk.
182 *
183 * Attribute functions don't use CURL, so they do much simpler in-memory
184 * buffering. Queries also don't use CURL, but the MHD POST interface
185 * introduces some of its own complexity so see below for that.
186 **********/
187
188 static void
189 simple_closer (void *ctx)
190 {
191 my_state *ms = ctx;
192
193 DPRINTF("%s: cleaning up\n",__func__);
194 free_ms(ms);
195 }
196
197 static void
198 child_closer (void * ctx)
199 {
200 pipe_private *pp = ctx;
201
202 DPRINTF("in %s\n",__func__);
203
204 free(pp);
205 }
206
207 /* Invoked from MHD. */
208 static int
209 proxy_get_cons (void *ctx, uint64_t pos, char *buf, int max)
210 {
211 pipe_private *pp = ctx;
212 pipe_shared *ps = pp->shared;
213 my_state *ms = ps->owner;
214 int done;
215 void *child_res;
216
217 (void)pos;
218
219 DPRINTF("consumer asked to read %d\n",max);
220
221 if (pipe_cons_wait(pp)) {
222 DPRINTF("consumer offset %zu into %zu\n",
223 pp->offset, ps->data_len);
224 done = ps->data_len - pp->offset;
225 if (done > max) {
226 done = max;
227 }
228 memcpy(buf,ps->data_ptr+pp->offset,done);
229 pp->offset += done;
230 DPRINTF("consumer copied %d, new offset %zu\n",
231 done, pp->offset);
232 if (pp->offset == ps->data_len) {
233 DPRINTF("consumer finished chunk\n");
234 pipe_cons_signal(pp, 0);
235 }
236 }
237 else {
238 done = -1;
239 }
240
241 if (done == (-1)) {
242 child_res = NULL;
243 pthread_join(ms->backend_th,&child_res);
244 if (child_res == THREAD_FAILED) {
245 ms->rc = MHD_HTTP_INTERNAL_SERVER_ERROR;
246 }
247 if (ms->from_master) {
248 pthread_join(ms->cache_th,NULL);
249 /* TBD: do something about cache failure? */
250 }
251 free_ms(ms);
252 }
253
254 return done;
255 }
256
257 static int
258 proxy_get_data (void *cctx, struct MHD_Connection *conn, const char *url,
259 const char *method, const char *version, const char *data,
260 size_t *data_size, void **rctx)
261 {
262 struct MHD_Response *resp;
263 my_state *ms = *rctx;
264 pipe_private *pp;
265 pipe_private *pp2;
266 char *my_etag;
267 const char *user_etag;
268
269 (void)cctx;
270 (void)method;
271 (void)version;
272 (void)data;
273 (void)data_size;
274
275 DPRINTF("PROXY GET DATA %s\n",url);
276
277 ms->url = strdup(url);
278 if (!ms->url) {
279 return MHD_NO;
280 }
281 ms->cleanup |= CLEANUP_URL;
282
283 my_etag = meta_has_copy(ms->bucket,ms->key,me);
284 if (my_etag) {
285 user_etag = MHD_lookup_connection_value(
286 conn, MHD_HEADER_KIND, "If-None-Match");
287 if (user_etag && !strcmp(user_etag,my_etag)) {
288 DPRINTF("ETag match!\n");
289 free(my_etag);
290 resp = MHD_create_response_from_data(0,NULL,
291 MHD_NO,MHD_NO);
292 MHD_queue_response(conn,MHD_HTTP_NOT_MODIFIED,resp);
293 MHD_destroy_response(resp);
294 return MHD_YES;
295 }
296 free(my_etag);
297 ms->from_master = 0;
298 }
299 else {
300 DPRINTF("%s/%s not found locally\n",ms->bucket,ms->key);
301 if (!master_host) {
302 DPRINTF(" that means it doesn't exist\n");
303 resp = MHD_create_response_from_data(0,NULL,
304 MHD_NO,MHD_NO);
305 MHD_queue_response(conn,MHD_HTTP_NOT_FOUND,resp);
306 MHD_destroy_response(resp);
307 free_ms(ms);
308 return MHD_YES;
309 }
310 DPRINTF(" will fetch from %s:%u\n", master_host,master_port);
311 ms->from_master = 1;
312 }
313
314 pipe_init_shared(&ms->pipe,ms,ms->from_master+1);
315 pp = pipe_init_private(&ms->pipe);
316 if (!pp) {
317 return MHD_NO;
318 }
319 /* Master is always assumed to be CURL (i.e. our own protocol) */
320 if (ms->from_master) {
321 pthread_create(&ms->backend_th,NULL,
322 curl_func_tbl.get_child_func,ms);
323 }
324 else {
325 pthread_create(&ms->backend_th,NULL,
326 main_func_tbl->get_child_func,ms);
327 }
328 /* TBD: check return value */
329
330 if (ms->from_master) {
331 pp2 = pipe_init_private(&ms->pipe);
332 if (!pp2) {
333 return MHD_NO;
334 }
335 pthread_create(&ms->cache_th,NULL,
336 main_func_tbl->cache_child_func,pp2);
337 /* TBD: check return value */
338 }
339 else {
340 pp2 = NULL;
341 }
342
343 resp = MHD_create_response_from_callback(
344 MHD_SIZE_UNKNOWN, 65536, proxy_get_cons, pp, child_closer);
345 if (!resp) {
346 fprintf(stderr,"MHD_crfc failed\n");
347 if (pp2) {
348 /* TBD: terminate thread */
349 free(pp2);
350 }
351 child_closer(pp);
352 return MHD_NO;
353 }
354 MHD_queue_response(conn,ms->rc,resp);
355 MHD_destroy_response(resp);
356
357 return MHD_YES;
358 }
359
360 static void
361 recheck_replication (my_state * ms, char *policy)
362 {
363 int rc;
364 int free_it = FALSE;
365 char fixed[MAX_FIELD_LEN];
366
367 if (is_reserved(ms->key,reserved_name)) {
368 DPRINTF("declining to replicate reserved object %s\n",ms->key);
369 return;
370 }
371
372 if (!policy && ms->dict) {
373 DPRINTF("using new policy for %s/%s\n",ms->bucket,ms->key);
374 policy = g_hash_table_lookup(ms->dict,"_policy");
375 }
376
377 if (!policy) {
378 /* If we get a policy here or below, we have to free it. */
379 free_it = TRUE;
380 DPRINTF("fetching policy for %s/%s\n",ms->bucket,ms->key);
381 rc = meta_get_value(ms->bucket,ms->key, "_policy", &policy);
382 }
383
384 if (!policy) {
385 DPRINTF(" inheriting policy from %s\n",ms->bucket);
386 rc = meta_get_value(ms->bucket,
387 "_default", "_policy", &policy);
388 }
389
390 if (policy) {
391 DPRINTF(" implementing policy %s\n",policy);
392 /*
393 * Can't use ms->url here because it might be a bucket POST
394 * and in that case ms->url points to the bucket.
395 */
396 snprintf(fixed,sizeof(fixed),"%s/%s",ms->bucket,ms->key);
397 replicate(fixed,0,policy);
398 if (free_it) {
399 free(policy);
400 }
401 }
402 else {
403 DPRINTF(" could not find a policy anywhere!\n");
404 }
405 }
406
407 static int
408 proxy_put_data (void *cctx, struct MHD_Connection *conn, const char *url,
409 const char *method, const char *version, const char *data,
410 size_t *data_size, void **rctx)
411 {
412 struct MHD_Response *resp;
413 my_state *ms = *rctx;
414 pipe_private *pp;
415 int rc;
416 char *etag = NULL;
417 void *child_res;
418
419 (void)cctx;
420 (void)method;
421 (void)version;
422
423 DPRINTF("PROXY PUT DATA %s (%zu)\n",url,*data_size);
424
425 if (ms->state == MS_NEW) {
426 if (!validate_put(conn) || !validate_url(url)) {
427 DPRINTF("rejecting %s\n",url);
428 resp = MHD_create_response_from_data(0,NULL,
429 MHD_NO,MHD_NO);
430 if (!resp) {
431 return MHD_NO;
432 }
433 MHD_queue_response(conn,MHD_HTTP_FORBIDDEN,resp);
434 MHD_destroy_response(resp);
435 return MHD_YES;
436 }
437 ms->state = MS_NORMAL;
438 ms->url = strdup(url);
439 if (!ms->url) {
440 return MHD_NO;
441 }
442 ms->cleanup |= CLEANUP_URL;
443 ms->size = 0;
444 pipe_init_shared(&ms->pipe,ms,1);
445 pp = pipe_init_private(&ms->pipe);
446 if (!pp) {
447 return MHD_NO;
448 }
449 pthread_create(&ms->backend_th,NULL,
450 main_func_tbl->put_child_func,pp);
451 /* TBD: check return value */
452
453 /*
454 * Do the initial handshake with children. If we return from
455 * this callback without an error response, Microhttpd posts
456 * the "100 Continue" header and the client starts sending
457 * the data. We must report errors here or forever keep
458 * out peace.
459 */
460 rc = pipe_prod_wait_init(&ms->pipe);
461 if (rc < 0) {
462 DPRINTF("producer wait failed\n");
463 resp = MHD_create_response_from_data(0,NULL,
464 MHD_NO,MHD_NO);
465 if (!resp) {
466 return MHD_NO;
467 }
468 MHD_queue_response(conn,MHD_HTTP_INTERNAL_SERVER_ERROR,
469 resp);
470 MHD_destroy_response(resp);
471 } else if (rc > 0) {
472 /*
473 * Note that we fail here even if 1 of N replicas fail.
474 * Might want to fix this when we start looping over
475 * pipe_init_private() above.
476 */
477 DPRINTF("producer replicas failed (%u of %u)\n",
478 rc, ms->pipe.cons_total);
479 resp = MHD_create_response_from_data(0,NULL,
480 MHD_NO,MHD_NO);
481 if (!resp) {
482 return MHD_NO;
483 }
484 MHD_queue_response(conn,MHD_HTTP_INTERNAL_SERVER_ERROR,
485 resp);
486 MHD_destroy_response(resp);
487 } else {
488 DPRINTF("producer proceeding\n");
489 }
490 }
491 else if (*data_size) {
492 pipe_prod_signal(&ms->pipe,(void *)data,*data_size);
493 ms->size += *data_size;
494 DPRINTF("producer chunk finished\n");
495 *data_size = 0;
496 }
497 else {
498 pipe_prod_finish(&ms->pipe);
499 pthread_join(ms->backend_th,&child_res);
500 if (child_res == THREAD_FAILED) {
501 DPRINTF("thread failed\n");
502 rc = MHD_HTTP_INTERNAL_SERVER_ERROR;
503 }
504 else if (ms->pipe.cons_error == ms->pipe.cons_total) {
505 DPRINTF("all %u consumers failed\n",
506 ms->pipe.cons_error);
507 rc = MHD_HTTP_INTERNAL_SERVER_ERROR;
508 }
509 else {
510 if (master_host) {
511 meta_got_copy(ms->bucket,ms->key,me);
512 etag = NULL;
513 }
514 else {
515 etag = meta_did_put(ms->bucket,ms->key,me,
516 ms->size);
517 }
518 DPRINTF("rereplicate (obj PUT)\n");
519 recheck_replication(ms,NULL);
520 rc = MHD_HTTP_OK;
521 }
522 free_ms(ms);
523 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
524 if (!resp) {
525 free(etag);
526 return MHD_NO;
527 }
528 if (etag) {
529 MHD_add_response_header(resp,"ETag",etag);
530 free(etag);
531 }
532 MHD_queue_response(conn,rc,resp);
533 MHD_destroy_response(resp);
534 }
535
536 return MHD_YES;
537 }
538
539 static int
540 proxy_get_attr (void *cctx, struct MHD_Connection *conn, const char *url,
541 const char *method, const char *version, const char *data,
542 size_t *data_size, void **rctx)
543 {
544 struct MHD_Response *resp;
545 char *fixed;
546 my_state *ms = *rctx;
547 int rc = MHD_HTTP_NOT_FOUND;
548
549 (void)cctx;
550 (void)method;
551 (void)version;
552 (void)data;
553 (void)data_size;
554
555 DPRINTF("PROXY GET ATTR %s\n",url);
556
557 if (meta_get_value(ms->bucket,ms->key,ms->attr,&fixed) == 0) {
558 resp = MHD_create_response_from_data(strlen(fixed),fixed,
559 MHD_YES,MHD_NO);
560 rc = MHD_HTTP_OK;
561 }
562 else {
563 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
564 }
565 if (!resp) {
566 return MHD_NO;
567 }
568 MHD_queue_response(conn,rc,resp);
569 MHD_destroy_response(resp);
570
571 free_ms(ms);
572 return MHD_YES;
573 }
574
575 static int
576 proxy_put_attr (void *cctx, struct MHD_Connection *conn, const char *url,
577 const char *method, const char *version, const char *data,
578 size_t *data_size, void **rctx)
579 {
580 struct MHD_Response *resp;
581 my_state *ms = *rctx;
582 const char *attrval;
583 int send_resp = 0;
584
585 (void)cctx;
586 (void)method;
587 (void)version;
588
589 DPRINTF("PROXY PUT ATTR %s (%zu)\n",url,*data_size);
590
591 if (ms->state == MS_NEW) {
592 ms->state = MS_NORMAL;
593 ms->url = strdup(url);
594 if (!ms->url) {
595 return MHD_NO;
596 }
597 ms->cleanup |= CLEANUP_URL;
598 attrval = MHD_lookup_connection_value(conn,MHD_HEADER_KIND,
599 "X-redhat-value");
600 if (attrval) {
601 meta_set_value(ms->bucket,ms->key,ms->attr,
602 (char *)attrval);
603 send_resp = 1;
604 }
605 }
606 else if (*data_size) {
607 if (ms->pipe.data_len) {
608 ms->pipe.data_len += *data_size;
609 char *p = realloc(ms->pipe.data_ptr,ms->pipe.data_len);
610 if (!p) {
611 return MHD_NO;
612 }
613 ms->pipe.data_ptr = p;
614 }
615 else {
616 ms->pipe.data_len = *data_size + 1;
617 ms->pipe.data_ptr = malloc(ms->pipe.data_len);
618 if (!ms->pipe.data_ptr) {
619 return MHD_NO;
620 }
621 ((char *)ms->pipe.data_ptr)[0] = '\0';
622 ms->cleanup |= CLEANUP_BUF_PTR;
623 }
624 (void)strncat(ms->pipe.data_ptr,data,*data_size);
625 /* TBD: check return value */
626 *data_size = 0;
627 }
628 else {
629 if (!ms->pipe.data_ptr) {
630 return MHD_NO;
631 }
632 if (is_reserved(ms->attr,reserved_attr)) {
633 resp = MHD_create_response_from_data(
634 0,NULL,MHD_NO,MHD_NO);
635 if (!resp) {
636 return MHD_NO;
637 }
638 MHD_queue_response(conn,MHD_HTTP_BAD_REQUEST,
639 resp);
640 MHD_destroy_response(resp);
641 free_ms(ms);
642 return MHD_YES;
643 }
644 meta_set_value(ms->bucket,ms->key,ms->attr,ms->pipe.data_ptr);
645 /*
646 * We should always re-replicate, because the replication
647 * policy might refer to this attr.
648 */
649 DPRINTF("rereplicate (attr PUT)\n");
650 recheck_replication(ms,NULL);
651 free_ms(ms);
652 send_resp = 1;
653 }
654
655 if (send_resp) {
656 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
657 if (!resp) {
658 return MHD_NO;
659 }
660 MHD_queue_response(conn,MHD_HTTP_CREATED,resp);
661 MHD_destroy_response(resp);
662 /*
663 * TBD: check if the attribute was a replication policy, and
664 * start/stop replication activities as appropriate.
665 */
666 }
667
668 return MHD_YES;
669 }
670
671 /**********
672 * For queries, we have to deal with MHD's post-iterator interface (not
673 * quite the same as the data-iteration even though we use it that way) on
674 * one side, and a query-iterator interface on the other. Data on both
675 * sides could be quite large, so we can't just stick them in header lines.
676 * We do still buffer the query in memory, though. Once that's done, we do
677 * very simple parsing - it will be more complicated later - and create the
678 * query iterator. That's also driven by MHD, this time though the
679 * content-callback interface, and repeatedly calls in to the metadata
680 * module to fetch one object name at a time.
681 **********/
682
683 static int
684 query_iterator (void *ctx, enum MHD_ValueKind kind, const char *key,
685 const char *filename, const char *content_type,
686 const char *transfer_encoding, const char *data,
687 uint64_t off, size_t size)
688 {
689 (void)ctx;
690 (void)kind;
691 (void)key;
692 (void)filename;
693 (void)content_type;
694 (void)transfer_encoding;
695 (void)data;
696 (void)off;
697 (void)size;
698
699 /* We actually accumulate the data in proxy_query. */
700 return MHD_YES;
701 }
702
703 /* MHD reader function during queries. Return -1 for EOF. */
704 static int
705 proxy_query_func (void *ctx, uint64_t pos, char *buf, int max)
706 {
707 my_state *ms = ctx;
708 int len;
709 const char *accept_hdr;
710 char *bucket;
711 char *key;
712
713 (void)pos;
714
715 accept_hdr = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,
716 "Accept");
717
718 if (!ms->gen_ctx) {
719 ms->gen_ctx = tmpl_get_ctx(accept_hdr);
720 if (!ms->gen_ctx) {
721 return -1;
722 }
723 ms->cleanup |= CLEANUP_TMPL;
724 len = tmpl_obj_header(ms->gen_ctx);
725 if (!len) {
726 return -1;
727 }
728 if (len > max) {
729 len = max;
730 }
731 memcpy(buf,ms->gen_ctx->buf,len);
732 return len;
733 }
734
735 if (ms->gen_ctx == TMPL_CTX_DONE) {
736 return -1;
737 }
738
739 for(;;) {
740 if (!meta_query_next(ms->query,&bucket,&key)) {
741 break;
742 }
743 if (is_reserved(key,reserved_name)) {
744 continue;
745 }
746 len = tmpl_obj_entry(ms->gen_ctx,bucket,key);
747 if (!len) {
748 return -1;
749 }
750 if (len > max) {
751 len = max;
752 }
753 memcpy(buf,ms->gen_ctx->buf,len);
754 return len;
755 }
756
757 len = tmpl_obj_footer(ms->gen_ctx);
758 if (!len) {
759 return -1;
760 }
761 if (len > max) {
762 len = max;
763 }
764 memcpy(buf,ms->gen_ctx->buf,len);
765 free(ms->gen_ctx);
766 ms->cleanup &= ~CLEANUP_TMPL;
767 ms->gen_ctx = TMPL_CTX_DONE;
768 return len;
769 }
770
771 static int
772 proxy_query (void *cctx, struct MHD_Connection *conn, const char *url,
773 const char *method, const char *version, const char *data,
774 size_t *data_size, void **rctx)
775 {
776 struct MHD_Response *resp;
777 my_state *ms = *rctx;
778
779 (void)cctx;
780 (void)method;
781 (void)version;
782
783 DPRINTF("PROXY QUERY %s (%zu)\n",url,*data_size);
784
785 if (ms->state == MS_NEW) {
786 ms->state = MS_NORMAL;
787 ms->post = MHD_create_post_processor(conn,4096,
788 query_iterator,ms);
789 ms->cleanup |= CLEANUP_POST;
790 }
791 else if (*data_size) {
792 MHD_post_process(ms->post,data,*data_size);
793 if (ms->pipe.data_len) {
794 ms->pipe.data_len += *data_size;
795 char *p = realloc(ms->pipe.data_ptr,ms->pipe.data_len);
796 if (!p) {
797 return MHD_NO;
798 }
799 ms->pipe.data_ptr = p;
800 }
801 else {
802 ms->pipe.data_len = *data_size + 1;
803 ms->pipe.data_ptr = malloc(ms->pipe.data_len);
804 if (!ms->pipe.data_ptr) {
805 return MHD_NO;
806 }
807 ((char *)ms->pipe.data_ptr)[0] = '\0';
808 ms->cleanup |= CLEANUP_BUF_PTR;
809 }
810 (void)strncat(ms->pipe.data_ptr,data,*data_size);
811 /* TBD: check return value */
812 *data_size = 0;
813 }
814 else {
815 if (!ms->pipe.data_ptr) {
816 return MHD_NO;
817 }
818 ms->query = meta_query_new(ms->bucket,NULL,ms->pipe.data_ptr);
819 ms->cleanup |= CLEANUP_QUERY;
820 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
821 65536, proxy_query_func, ms, simple_closer);
822 if (!resp) {
823 fprintf(stderr,"MHD_crfc failed\n");
824 simple_closer(ms);
825 return MHD_NO;
826 }
827 MHD_queue_response(conn,MHD_HTTP_OK,resp);
828 MHD_destroy_response(resp);
829 free_ms(ms);
830 }
831
832 return MHD_YES;
833 }
834
835 static int
836 proxy_list_objs (void *cctx, struct MHD_Connection *conn, const char *url,
837 const char *method, const char *version, const char *data,
838 size_t *data_size, void **rctx)
839 {
840 my_state *ms = *rctx;
841 struct MHD_Response *resp;
842
843 (void)cctx;
844 (void)url;
845 (void)method;
846 (void)version;
847 (void)data;
848 (void)data_size;
849
850 ms->query = meta_query_new((char *)ms->bucket,NULL,NULL);
851 ms->cleanup |= CLEANUP_QUERY;
852
853 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
854 65536, proxy_query_func, ms, simple_closer);
855 if (!resp) {
856 fprintf(stderr,"MHD_crfc failed\n");
857 simple_closer(ms);
858 return MHD_NO;
859 }
860
861 MHD_queue_response(conn,MHD_HTTP_OK,resp);
862 MHD_destroy_response(resp);
863 return MHD_YES;
864 }
865
866 static int
867 proxy_delete (void *cctx, struct MHD_Connection *conn, const char *url,
868 const char *method, const char *version, const char *data,
869 size_t *data_size, void **rctx)
870 {
871 my_state *ms = *rctx;
872 struct MHD_Response *resp;
873 char *copied_url;
874 char *bucket;
875 char *key;
876 char *stctx = NULL;
877 int rc;
878
879 (void)cctx;
880 (void)method;
881 (void)version;
882 (void)data;
883 (void)data_size;
884
885 DPRINTF("PROXY DELETE %s\n",url);
886
887 rc = main_func_tbl->delete_func(ms->bucket,ms->key,url);
888 if (rc == MHD_HTTP_OK) {
889 copied_url = strdup(url);
890 assert (copied_url);
891 bucket = strtok_r(copied_url,"/",&stctx);
892 key = strtok_r(NULL,"/",&stctx);
893 meta_delete(bucket,key);
894 free(copied_url);
895 replicate_delete(url);
896 }
897
898 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
899 if (!resp) {
900 return MHD_NO;
901 }
902 MHD_queue_response(conn,rc,resp);
903 MHD_destroy_response(resp);
904
905 free_ms(ms);
906 return MHD_YES;
907 }
908
909 /* TBD: get actual bucket list */
910 typedef struct {
911 char *rel;
912 char *link;
913 } fake_bucket_t;
914
915 static const fake_bucket_t fake_bucket_list[] = {
916 { "bucket_factory", "_new" },
917 { "provider_list", "_providers" },
918 };
919
920 static int
921 root_blob_generator (void *ctx, uint64_t pos, char *buf, int max)
922 {
923 my_state *ms = ctx;
924 const fake_bucket_t *fb;
925 int len;
926 const char *accept_hdr;
927 const char *host;
928 char *bucket;
929 char *key;
930
931 (void)pos;
932
933 accept_hdr = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,
934 "Accept");
935 host = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,"Host");
936
937 if (!ms->gen_ctx) {
938 ms->gen_ctx = tmpl_get_ctx(accept_hdr);
939 if (!ms->gen_ctx) {
940 return -1;
941 }
942 ms->cleanup |= CLEANUP_TMPL;
943 ms->gen_ctx->base = host;
944 len = tmpl_root_header(ms->gen_ctx,"image_warehouse","1.0");
945 if (!len) {
946 return -1;
947 }
948 if (len > max) {
949 len = max;
950 }
951 memcpy(buf,ms->gen_ctx->buf,len);
952 return len;
953 }
954
955 if (ms->gen_ctx == TMPL_CTX_DONE) {
956 return -1;
957 }
958
959 if (ms->gen_ctx->index < ARRAY_SIZE(fake_bucket_list)) {
960 fb = fake_bucket_list + ms->gen_ctx->index;
961 len = tmpl_root_entry(ms->gen_ctx,fb->rel,fb->link);
962 if (!len) {
963 return -1;
964 }
965 if (len > max) {
966 len = max;
967 }
968 memcpy(buf,ms->gen_ctx->buf,len);
969 return len;
970 }
971
972 if (meta_query_next(ms->query,&bucket,&key)) {
973 len = tmpl_root_entry(ms->gen_ctx,"bucket",bucket);
974 if (!len) {
975 return -1;
976 }
977 if (len > max) {
978 len = max;
979 }
980 memcpy(buf,ms->gen_ctx->buf,len);
981 return len;
982 }
983
984 len = tmpl_root_footer(ms->gen_ctx);
985 if (!len) {
986 return -1;
987 }
988 if (len > max) {
989 len = max;
990 }
991 memcpy(buf,ms->gen_ctx->buf,len);
992 free(ms->gen_ctx);
993 ms->cleanup &= ~CLEANUP_TMPL;
994 ms->gen_ctx = TMPL_CTX_DONE;
995 return len;
996 }
997
998 static int
999 proxy_api_root (void *cctx, struct MHD_Connection *conn, const char *url,
1000 const char *method, const char *version, const char *data,
1001 size_t *data_size, void **rctx)
1002 {
1003 struct MHD_Response *resp = NULL;
1004 unsigned int rc = MHD_HTTP_OK;
1005 my_state *ms = *rctx;
1006
1007 (void)cctx;
1008 (void)method;
1009 (void)version;
1010 (void)data;
1011
1012 DPRINTF("PROXY API ROOT (%s, %zu)\n",url,*data_size);
1013
1014 ms->query = meta_query_new(NULL,"_default",NULL);
1015 if (!ms->query) {
1016 free_ms(ms);
1017 return MHD_NO;
1018 }
1019 ms->cleanup |= CLEANUP_QUERY;
1020
1021 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
1022 65536, root_blob_generator, ms, simple_closer);
1023 if (!resp) {
1024 return MHD_NO;
1025 }
1026 MHD_queue_response(conn,rc,resp);
1027 MHD_destroy_response(resp);
1028
1029 return MHD_YES;
1030
1031 }
1032
1033 static int
1034 post_iterator (void *ctx, enum MHD_ValueKind kind, const char *key,
1035 const char *filename, const char *content_type,
1036 const char *transfer_encoding, const char *data,
1037 uint64_t off, size_t size)
1038 {
1039 char *old_val;
1040 size_t old_len;
1041 char *new_val;
1042
1043 (void)kind;
1044 (void)filename;
1045 (void)content_type;
1046 (void)transfer_encoding;
1047 (void)off;
1048
1049 printf("adding %s, size=%zu\n",key,size);
1050
1051 // TBD: don't assume that values are null-terminated strings
1052 old_val = g_hash_table_lookup(ctx,key);
1053 if (old_val) {
1054 old_len = strlen(old_val);
1055 new_val = malloc(old_len+size+1);
1056 if (!new_val) {
1057 return MHD_NO;
1058 }
1059 memcpy(new_val,old_val,old_len);
1060 memcpy(new_val+old_len,data,size);
1061 new_val[old_len+size] = '\0';
1062 }
1063 else {
1064 new_val = malloc(size+1);
1065 if (!new_val) {
1066 return MHD_NO;
1067 }
1068 memcpy(new_val,data,size);
1069 new_val[size] = '\0';
1070 }
1071
1072 g_hash_table_insert(ctx,strdup(key),new_val);
1073 /* TBD: check return value for strdups (none avail for insert) */
1074 return MHD_YES;
1075 }
1076
1077 /* Returns TRUE if we found an *invalid* key. */
1078 static gboolean
1079 post_find (gpointer key, gpointer value, gpointer ctx)
1080 {
1081 (void)value;
1082 (void)ctx;
1083
1084 if (!is_reserved(key,reserved_attr)) {
1085 return FALSE;
1086 }
1087
1088 DPRINTF("bad attr %s\n", (char *)key);
1089 return TRUE;
1090 }
1091
1092 static void
1093 post_foreach (gpointer key, gpointer value, gpointer ctx)
1094 {
1095 my_state *ms = ctx;
1096
1097 DPRINTF("setting %s = %s for %s/%s\n",(char *)key, (char *)value,
1098 ms->bucket,ms->key);
1099 meta_set_value(ms->bucket,ms->key,key,value);
1100 }
1101
1102 static int
1103 create_bucket (char *name)
1104 {
1105 int rc;
1106
1107 if (is_reserved(name,reserved_name)) {
1108 return MHD_HTTP_BAD_REQUEST;
1109 }
1110
1111 rc = main_func_tbl->bcreate_func(name);
1112 if (rc == MHD_HTTP_OK) {
1113 if (meta_set_value(name,"_default", "_policy","0") != 0) {
1114 DPRINTF("default-policy " "create failed\n");
1115 /* Non-fatal. */
1116 }
1117 DPRINTF("created bucket %s\n",name);
1118 /*
1119 * There's not a whole lot to do about bucket-creation
1120 * failures on replicas, other than to report them, unless
1121 * we adopt an "all or nothing" approach and unwind the
1122 * create on the primary as well. Then what if that fails?
1123 * It's just one example of the general "fewer replicas
1124 * than desired" distributed-system problem, not worth a
1125 * point solution here/now. Revisit when we have a more
1126 * general replica-repair policy/system in place.
1127 */
1128 replicate_bcreate(name);
1129 }
1130
1131 return rc;
1132 }
1133
1134 static int
1135 proxy_bucket_post (void *cctx, struct MHD_Connection *conn, const char *url,
1136 const char *method, const char *version, const char *data,
1137 size_t *data_size, void **rctx)
1138 {
1139 struct MHD_Response *resp;
1140 my_state *ms = *rctx;
1141 int rc;
1142 char *key;
1143
1144 (void)cctx;
1145 (void)method;
1146 (void)version;
1147
1148 DPRINTF("PROXY POST (%s, %zu)\n",url,*data_size);
1149
1150 if (ms->state == MS_NEW) {
1151 ms->state = MS_NORMAL;
1152 ms->url = (char *)url;
1153 ms->dict = g_hash_table_new_full(
1154 g_str_hash,g_str_equal,free,free);
1155 ms->cleanup |= CLEANUP_DICT;
1156 ms->post = MHD_create_post_processor(conn,4096,
1157 post_iterator,ms->dict);
1158 ms->cleanup |= CLEANUP_POST;
1159 }
1160 else if (*data_size) {
1161 MHD_post_process(ms->post,data,*data_size);
1162 *data_size = 0;
1163 }
1164 else {
1165 rc = MHD_HTTP_BAD_REQUEST;
1166 key = g_hash_table_lookup(ms->dict,"key");
1167 if (key) {
1168 strncpy(ms->key,key,MAX_FIELD_LEN-1);
1169 g_hash_table_remove(ms->dict,"key");
1170 if (!g_hash_table_find(ms->dict,post_find,ms)) {
1171 g_hash_table_foreach(ms->dict,post_foreach,ms);
1172 DPRINTF("rereplicate (bucket POST)\n");
1173 recheck_replication(ms,NULL);
1174 rc = MHD_HTTP_OK;
1175 }
1176 }
1177 else if (!strcmp(ms->bucket,"_new")) {
1178 key = g_hash_table_lookup(ms->dict,"name");
1179 if (key != NULL) {
1180 rc = create_bucket(key);
1181 }
1182 }
1183 else {
1184 DPRINTF("A parameter is MISSING (fail)\n");
1185 }
1186 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1187 if (!resp) {
1188 fprintf(stderr,"MHD_crfd failed\n");
1189 return MHD_NO;
1190 }
1191 MHD_queue_response(conn,rc,resp);
1192 MHD_destroy_response(resp);
1193 free_ms(ms);
1194 }
1195
1196 return MHD_YES;
1197 }
1198
1199 static int
1200 check_location (my_state *ms)
1201 {
1202 char *loc = g_hash_table_lookup(ms->dict,"depot");
1203
1204 if (!loc) {
1205 DPRINTF("missing loc on check for %s/%s\n",ms->bucket,ms->key);
1206 return MHD_HTTP_BAD_REQUEST;
1207 }
1208
1209 if (!meta_has_copy(ms->bucket,ms->key,loc)) {
1210 DPRINTF("did not find %s/%s at %s\n",ms->bucket,ms->key,loc);
1211 return MHD_HTTP_NOT_FOUND;
1212 }
1213
1214 /* TBD: meta_has_copy returns an etag which we should check */
1215 DPRINTF("found %s/%s at %s\n",ms->bucket,ms->key,loc);
1216 return MHD_HTTP_OK;
1217 }
1218
1219 static int
1220 register_image (my_state *ms)
1221 {
1222 char *site;
1223 provider_t prov;
1224 int i;
1225 char *next;
1226
1227 site = g_hash_table_lookup(ms->dict,"site");
1228 if (!site) {
1229 printf("site MISSING\n");
1230 return MHD_HTTP_BAD_REQUEST;
1231 }
1232
1233 next = index(site,':');
1234 if (next) {
1235 *(next++) = '\0';
1236 }
1237
1238 for (i = 0; get_provider(i,&prov); ++i) {
1239 if (strcmp(prov.name,site)) {
1240 continue;
1241 }
1242 return prov.func_tbl->register_func(ms,&prov,next,ms->dict);
1243 }
1244
1245 DPRINTF("site %s not found\n",site);
1246 return MHD_HTTP_BAD_REQUEST;
1247 }
1248
1249 static int
1250 proxy_object_post (void *cctx, struct MHD_Connection *conn, const char *url,
1251 const char *method, const char *version, const char *data,
1252 size_t *data_size, void **rctx)
1253 {
1254 struct MHD_Response *resp;
1255 my_state *ms = *rctx;
1256 int rc;
1257 char *op;
1258
1259 (void)cctx;
1260 (void)method;
1261 (void)version;
1262
1263 DPRINTF("PROXY POST (%s, %zu)\n",url,*data_size);
1264
1265 if (ms->state == MS_NEW) {
1266 ms->state = MS_NORMAL;
1267 ms->url = (char *)url;
1268 ms->dict = g_hash_table_new_full(
1269 g_str_hash,g_str_equal,free,free);
1270 ms->cleanup |= CLEANUP_DICT;
1271 ms->post = MHD_create_post_processor(conn,4096,
1272 post_iterator,ms->dict);
1273 ms->cleanup |= CLEANUP_POST;
1274 }
1275 else if (*data_size) {
1276 MHD_post_process(ms->post,data,*data_size);
1277 *data_size = 0;
1278 }
1279 else {
1280 rc = MHD_HTTP_BAD_REQUEST;
1281 if (!g_hash_table_find(ms->dict,post_find,ms)) {
1282 op = g_hash_table_lookup(ms->dict,"op");
1283 if (op) {
1284 if (!strcmp(op,"push")) {
1285 DPRINTF("rereplicate (obj POST)\n");
1286 recheck_replication(ms,NULL);
1287 rc = MHD_HTTP_OK;
1288 }
1289 else if (!strcmp(op,"check")) {
1290 rc = check_location(ms);
1291 }
1292 else if (!strcmp(op,"register")) {
1293 rc = register_image(ms);
1294 }
1295 else {
1296 DPRINTF("unknown op %s for %s/%s\n",
1297 op, ms->bucket, ms->key);
1298 }
1299 }
1300 else {
1301 DPRINTF("op is MISSING (fail)\n");
1302 }
1303 }
1304 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1305 if (!resp) {
1306 fprintf(stderr,"MHD_crfd failed\n");
1307 return MHD_NO;
1308 }
1309 MHD_queue_response(conn,rc,resp);
1310 MHD_destroy_response(resp);
1311 free_ms(ms);
1312 }
1313
1314 return MHD_YES;
1315
1316 }
1317
1318
1319 static int
1320 prov_list_generator (void *ctx, uint64_t pos, char *buf, int max)
1321 {
1322 my_state *ms = ctx;
1323 int len;
1324 provider_t prov;
1325 const char *accept_hdr;
1326
1327 (void)pos;
1328
1329 accept_hdr = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,
1330 "Accept");
1331
1332 if (!ms->gen_ctx) {
1333 ms->gen_ctx = tmpl_get_ctx(accept_hdr);
1334 if (!ms->gen_ctx) {
1335 return -1;
1336 }
1337 ms->cleanup |= CLEANUP_TMPL;
1338 len = tmpl_prov_header(ms->gen_ctx);
1339 if (!len) {
1340 return -1;
1341 }
1342 if (len > max) {
1343 len = max;
1344 }
1345 memcpy(buf,ms->gen_ctx->buf,len);
1346 return len;
1347 }
1348
1349 if (ms->gen_ctx == TMPL_CTX_DONE) {
1350 return -1;
1351 }
1352
1353 if (get_provider(ms->gen_ctx->index,&prov)) {
1354 len = tmpl_prov_entry(ms->gen_ctx,prov.name,prov.type,
1355 prov.host, prov.port, prov.username, prov.password);
1356 if (!len) {
1357 return -1;
1358 }
1359 if (len > max) {
1360 len = max;
1361 }
1362 memcpy(buf,ms->gen_ctx->buf,len);
1363 return len;
1364 }
1365
1366 len = tmpl_prov_footer(ms->gen_ctx);
1367 if (!len) {
1368 return -1;
1369 }
1370 if (len > max) {
1371 len = max;
1372 }
1373 memcpy(buf,ms->gen_ctx->buf,len);
1374 free(ms->gen_ctx);
1375 ms->cleanup &= ~CLEANUP_TMPL;
1376 ms->gen_ctx = TMPL_CTX_DONE;
1377 return len;
1378 }
1379
1380 static int
1381 proxy_list_provs (void *cctx, struct MHD_Connection *conn, const char *url,
1382 const char *method, const char *version, const char *data,
1383 size_t *data_size, void **rctx)
1384 {
1385 struct MHD_Response *resp;
1386 my_state *ms = *rctx;
1387
1388 (void)cctx;
1389 (void)url;
1390 (void)method;
1391 (void)version;
1392 (void)data;
1393 (void)data_size;
1394
1395 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
1396 65536, prov_list_generator, ms, simple_closer);
1397 if (!resp) {
1398 fprintf(stderr,"MHD_crfd failed\n");
1399 simple_closer(ms);
1400 return MHD_NO;
1401 }
1402 MHD_queue_response(conn,MHD_HTTP_OK,resp);
1403 MHD_destroy_response(resp);
1404
1405 return MHD_YES;
1406 }
1407
1408 static int
1409 prov_iterator (void *ctx, enum MHD_ValueKind kind, const char *key,
1410 const char *filename, const char *content_type,
1411 const char *transfer_encoding, const char *data,
1412 uint64_t off, size_t size)
1413 {
1414 (void)kind;
1415 (void)filename;
1416 (void)content_type;
1417 (void)transfer_encoding;
1418 (void)off;
1419
1420 g_hash_table_insert(ctx,strdup(key),strndup(data,size));
1421 /* TBD: check return value for strdups (none avail for insert) */
1422 return MHD_YES;
1423 }
1424
1425
1426 static int
1427 proxy_update_prov (void *cctx, struct MHD_Connection *conn, const char *url,
1428 const char *method, const char *version, const char *data,
1429 size_t *data_size, void **rctx)
1430 {
1431 struct MHD_Response *resp;
1432 my_state *ms = *rctx;
1433 int rc;
1434 char *provider;
1435 char *username;
1436 char *password;
1437
1438 (void)cctx;
1439 (void)method;
1440 (void)version;
1441
1442 if (ms->state == MS_NEW) {
1443 ms->state = MS_NORMAL;
1444 ms->url = (char *)url;
1445 ms->dict = g_hash_table_new_full(
1446 g_str_hash,g_str_equal,free,free);
1447 ms->cleanup |= CLEANUP_DICT;
1448 ms->post = MHD_create_post_processor(conn,4096,
1449 prov_iterator,ms->dict);
1450 ms->cleanup |= CLEANUP_POST;
1451 }
1452 else if (*data_size) {
1453 MHD_post_process(ms->post,data,*data_size);
1454 *data_size = 0;
1455 }
1456 else {
1457 rc = MHD_HTTP_BAD_REQUEST;
1458 provider = g_hash_table_lookup(ms->dict,"provider");
1459 username = g_hash_table_lookup(ms->dict,"username");
1460 password = g_hash_table_lookup(ms->dict,"password");
1461 if (provider && username && password) {
1462 update_provider(provider,username,password);
1463 rc = MHD_HTTP_OK;
1464 }
1465 else {
1466 DPRINTF("provider/username/password MISSING\n");
1467 }
1468 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1469 if (!resp) {
1470 fprintf(stderr,"MHD_crfd failed\n");
1471 return MHD_NO;
1472 }
1473 MHD_queue_response(conn,rc,resp);
1474 MHD_destroy_response(resp);
1475 free_ms(ms);
1476 }
1477
1478 return MHD_YES;
1479 }
1480
1481 static int
1482 proxy_create_bucket (void *cctx, struct MHD_Connection *conn, const char *url,
1483 const char *method, const char *version, const char *data,
1484 size_t *data_size, void **rctx)
1485 {
1486 struct MHD_Response *resp;
1487 my_state *ms = *rctx;
1488 int rc;
1489
1490 (void)cctx;
1491 (void)method;
1492 (void)version;
1493 (void)data;
1494 (void)data_size;
1495 (void)url;
1496
1497 /* curl -T moo.empty http://localhost:9090/_new by accident */
1498 rc = create_bucket(ms->bucket);
1499
1500 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1501 if (!resp) {
1502 fprintf(stderr,"MHD_crfd failed\n");
1503 return MHD_NO;
1504 }
1505 MHD_queue_response(conn,rc,resp);
1506 MHD_destroy_response(resp);
1507
1508 free_ms(ms);
1509 return MHD_YES;
1510 }
1511
1512 static const rule my_rules[] = {
1513 { /* get bucket list */
1514 "GET", URL_ROOT, proxy_api_root },
1515 { /* get object list */
1516 "GET", URL_BUCKET, proxy_list_objs },
1517 { /* create bucket */
1518 "PUT", URL_BUCKET, proxy_create_bucket },
1519 { /* get object data */
1520 "GET", URL_OBJECT, proxy_get_data },
1521 { /* get attribute data */
1522 "GET", URL_ATTR, proxy_get_attr },
1523 { /* put object data */
1524 "PUT", URL_OBJECT, proxy_put_data },
1525 { /* put attribute data */
1526 "PUT", URL_ATTR, proxy_put_attr },
1527 { /* create object and/or modify attributes */
1528 "POST", URL_BUCKET, proxy_bucket_post },
1529 { /* perform control operations on an object */
1530 "POST", URL_OBJECT, proxy_object_post },
1531 { /* query */
1532 "POST", URL_QUERY, proxy_query },
1533 { /* delete object */
1534 "DELETE", URL_OBJECT, proxy_delete },
1535 { /* delete attribute (TBD) */
1536 "DELETE", URL_ATTR, NULL },
1537 { /* get provider list */
1538 "GET", URL_PROVLIST, proxy_list_provs },
1539 { /* update a provider */
1540 "POST", URL_PROVLIST, proxy_update_prov },
1541 { NULL, 0, NULL }
1542 };
1543
1544 static url_type
1545 parse_url (const char *url, my_state *ms)
1546 {
1547 unsigned short esize;
1548 unsigned short eindex;
1549 char *parts[URL_INVAL];
1550
1551 if (strstr(url,"../")) {
1552 /* Naughty, naughty. Never a good reason to allow this. */
1553 DPRINTF("Rejecting ../ in path.\n");
1554 return URL_INVAL;
1555 }
1556
1557 eindex = URL_ROOT;
1558 parts[URL_BUCKET] = ms->bucket;
1559 parts[URL_OBJECT] = ms->key;
1560 parts[URL_ATTR] = ms->attr;
1561
1562 for (;;) {
1563 while (*url == '/') {
1564 ++url;
1565 }
1566
1567 if (!*url) {
1568 if (eindex == URL_BUCKET) {
1569 if (!strcmp(ms->bucket,"_providers")) {
1570 eindex = URL_PROVLIST;
1571 }
1572 }
1573 else if (eindex == URL_OBJECT) {
1574 if (!strcmp(ms->key,"_query")) {
1575 eindex = URL_QUERY;
1576 }
1577 }
1578 break;
1579 }
1580
1581 if (++eindex >= URL_INVAL) {
1582 return URL_INVAL;
1583 }
1584 esize = 0;
1585
1586 while (*url && (*url != '/')) {
1587 parts[eindex][esize++] = *(url++);
1588 if (esize >= MAX_FIELD_LEN) {
1589 return URL_INVAL;
1590 }
1591 }
1592 }
1593
1594 return eindex;
1595 }
1596
1597 static int
1598 access_handler (void *cctx, struct MHD_Connection *conn, const char *url,
1599 const char *method, const char *version, const char *data,
1600 size_t *data_size, void **rctx)
1601 {
1602 unsigned int i;
1603 url_type utype;
1604 struct MHD_Response *resp;
1605 my_state *ms = *rctx;
1606
1607 if (ms) {
1608 return ms->handler(cctx,conn,url,method,version,
1609 data,data_size,rctx);
1610 }
1611
1612 ms = calloc(sizeof(*ms), 1);
1613 if (!ms) {
1614 return MHD_NO;
1615 }
1616
1617 utype = parse_url(url,ms);
1618
1619 for (i = 0; my_rules[i].method; ++i) {
1620 if (utype != my_rules[i].utype) {
1621 continue;
1622 }
1623 if (strcmp(method,my_rules[i].method)) {
1624 continue;
1625 }
1626 if (!my_rules[i].handler) {
1627 break;
1628 }
1629 ms->handler = my_rules[i].handler;
1630 ms->state = MS_NEW;
1631 ms->fd = (-1);
1632 ms->url = NULL;
1633 ms->post = NULL;
1634 ms->conn = conn;
1635 *rctx = ms;
1636 return ms->handler(cctx,conn,url,method,version,
1637 data,data_size,rctx);
1638 }
1639
1640 if (!strcmp(method,"QUIT")) {
1641 (void)sem_post((sem_t *)cctx);
1642 return MHD_NO;
1643 }
1644
1645 fprintf(stderr,"bad request m=%s u=%s\n",method,url);
1646 free_ms(ms);
1647
1648 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1649 if (!resp) {
1650 return MHD_NO;
1651 }
1652 MHD_queue_response(conn,MHD_HTTP_NOT_FOUND,resp);
1653 MHD_destroy_response(resp);
1654 return MHD_YES;
1655 }
1656
1657 /* These enum values cannot possibly conflict with the option values
1658 ordinarily used by commands, including CHAR_MAX + 1, etc. Avoid
1659 CHAR_MIN - 1, as it may equal -1, the getopt end-of-options value. */
1660 enum
1661 {
1662 GETOPT_HELP_CHAR = (CHAR_MIN - 2),
1663 GETOPT_VERSION_CHAR = (CHAR_MIN - 3)
1664 };
1665
1666 static const struct option my_options[] = {
1667 { "config", required_argument, NULL, 'c' },
1668 { "db", required_argument, NULL, 'd' },
1669 { "master", required_argument, NULL, 'm' },
1670 { "port", required_argument, NULL, 'p' },
1671 { "verbose", no_argument, NULL, 'v' },
1672 { "version", no_argument, NULL, GETOPT_VERSION_CHAR },
1673 { "help", no_argument, NULL, GETOPT_HELP_CHAR },
1674 { NULL, 0, NULL, '\0' }
1675 };
1676
1677 static void
1678 usage (int status)
1679 {
1680 if (status != EXIT_SUCCESS)
1681 fprintf (stderr, _("Try `%s --help' for more information.\n"),
1682 program_name);
1683 else
1684 {
1685 printf (_("\
1686 Usage: %s [OPTION]\n\
1687 "),
1688 program_name);
1689 fputs (_("\
1690 Deltacloud image-warehouse daemon.\n\
1691 A configuration file must be specified.\n\
1692 \n\
1693 -c, --config=FILE config file [required]\n\
1694 -d, --db=HOST_PORT database server as ip[:port]\n\
1695 -m, --master=HOST_PORT master (upstream) server as ip[:port]\n\
1696 -p, --port=PORT alternate listen port (default 9090)\n\
1697 -v, --verbose verbose/debug output\n\
1698 \n\
1699 --help display this help and exit\n\
1700 --version output version information and exit\n\
1701 "), stdout);
1702 printf (_("\
1703 \n\
1704 Report %s bugs to %s.\n\
1705 "),
1706 program_name, PACKAGE_BUGREPORT);
1707 }
1708 exit (status);
1709 }
1710
1711 int
1712 main (int argc, char **argv)
1713 {
1714 struct MHD_Daemon *the_daemon;
1715 sem_t the_sem;
1716 char *stctx = NULL;
1717 char *port_tmp;
1718
1719 program_name = argv[0];
1720
1721 for (;;) switch (getopt_long(argc,argv,"c:d:m:p:v",my_options,NULL)) {
1722 case 'c':
1723 cfg_file = optarg;
1724 break;
1725 case 'd':
1726 assert (optarg);
1727 db_host = strtok_r(optarg,":",&stctx);
1728 port_tmp = strtok_r(NULL,":",&stctx);
1729 if (port_tmp) {
1730 db_port = (unsigned short)strtoul(port_tmp,NULL,10);
1731 }
1732 break;
1733 case 'm':
1734 assert (optarg);
1735 master_host = strtok_r(optarg,":",&stctx);
1736 port_tmp = strtok_r(NULL,":",&stctx);
1737 if (port_tmp) {
1738 master_port = (unsigned short)strtoul(port_tmp,NULL,10);
1739 }
1740 break;
1741 case 'p':
1742 my_port = (unsigned short)strtoul(optarg,NULL,10);
1743 break;
1744 case 'v':
1745 ++verbose;
1746 break;
1747 case GETOPT_HELP_CHAR:
1748 usage(EXIT_SUCCESS);
1749 break;
1750 case GETOPT_VERSION_CHAR:
1751 printf ("%s version %s\n", program_name, PACKAGE_VERSION);
1752 exit (EXIT_SUCCESS);
1753 break;
1754
1755 case -1:
1756 goto args_done;
1757 default:
1758 usage(EXIT_FAILURE);
1759 break;
1760 }
1761 args_done:
1762
1763 if (!cfg_file) {
1764 error (0, 0, "no configuration file specified");
1765 usage (EXIT_FAILURE);
1766 }
1767
1768 me = parse_config();
1769 if (!me) {
1770 fprintf(stderr,"could not parse %s\n",cfg_file);
1771 return !0;
1772 }
1773
1774 sem_init(&the_sem,0,0);
1775 if (proxy_host) {
1776 if (s3mode) {
1777 main_func_tbl = &s3_func_tbl;
1778 }
1779 else {
1780 main_func_tbl = &curl_func_tbl;
1781 }
1782 }
1783 else {
1784 main_func_tbl = &fs_func_tbl;
1785 }
1786
1787 if (verbose) {
1788 printf("primary store type is %s\n",main_func_tbl->name);
1789 if (master_host) {
1790 printf("operating as slave to %s:%u\n",
1791 master_host, master_port);
1792 }
1793 printf("db is at %s:%u\n",db_host,db_port);
1794 printf("will listen on port %u\n",my_port);
1795 printf("my location is \"%s\"\n",me);
1796 if (fflush(stdout) || ferror(stdout))
1797 error(EXIT_FAILURE, 0, "write failed");
1798 }
1799
1800 backend_init();
1801 main_func_tbl->init_func();
1802 meta_init();
1803 repl_init();
1804
1805 /*
1806 * Gotcha: if we don't set the connection memory limit explicitly,
1807 * the per-connection buffer for MHD will be smaller than that used
1808 * by CURL, so proxy_writefunc will never be able to do its job.
1809 */
1810 the_daemon = MHD_start_daemon(MY_MHD_FLAGS,
1811 my_port, NULL, NULL, &access_handler, &the_sem,
1812 MHD_OPTION_CONNECTION_MEMORY_LIMIT, (size_t)1048576,
1813 MHD_OPTION_END);
1814 if (!the_daemon) {
1815 fprintf(stderr,"Could not create daemon.\n");
1816 return !0;
1817 }
1818
1819 sem_wait(&the_sem);
1820 return 0;
1821 }
image warehouse daemon