?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Fix ms leaks when object not found, deleted.
[iwhd.git] / rest.c
blob66f91bba45eaf164e8bdcc10ecc299f74d26d3a3
1 /* Copyright (C) 2010 Free Software Foundation, 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 int fs_mode = 0;
73 unsigned short my_port = MY_PORT;
74 const char *program_name;
75
76 char *(reserved_name[]) = { "_default", "_query", "_new", NULL };
77 char *(reserved_attr[]) = { "bucket", "key", "date", "etag", "loc", NULL };
78
79 backend_func_tbl *main_func_tbl = &bad_func_tbl;
80
81 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 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 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 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 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 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 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 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 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 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 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 return MHD_YES;
572 }
573
574 int
575 proxy_put_attr (void *cctx, struct MHD_Connection *conn, const char *url,
576 const char *method, const char *version, const char *data,
577 size_t *data_size, void **rctx)
578 {
579 struct MHD_Response *resp;
580 my_state *ms = *rctx;
581 const char *attrval;
582 int send_resp = 0;
583
584 (void)cctx;
585 (void)method;
586 (void)version;
587
588 DPRINTF("PROXY PUT ATTR %s (%zu)\n",url,*data_size);
589
590 if (ms->state == MS_NEW) {
591 ms->state = MS_NORMAL;
592 ms->url = strdup(url);
593 if (!ms->url) {
594 return MHD_NO;
595 }
596 ms->cleanup |= CLEANUP_URL;
597 attrval = MHD_lookup_connection_value(conn,MHD_HEADER_KIND,
598 "X-redhat-value");
599 if (attrval) {
600 meta_set_value(ms->bucket,ms->key,ms->attr,
601 (char *)attrval);
602 send_resp = 1;
603 }
604 }
605 else if (*data_size) {
606 if (ms->pipe.data_len) {
607 ms->pipe.data_len += *data_size;
608 char *p = realloc(ms->pipe.data_ptr,ms->pipe.data_len);
609 if (!p) {
610 return MHD_NO;
611 }
612 ms->pipe.data_ptr = p;
613 }
614 else {
615 ms->pipe.data_len = *data_size + 1;
616 ms->pipe.data_ptr = malloc(ms->pipe.data_len);
617 if (!ms->pipe.data_ptr) {
618 return MHD_NO;
619 }
620 ((char *)ms->pipe.data_ptr)[0] = '\0';
621 ms->cleanup |= CLEANUP_BUF_PTR;
622 }
623 (void)strncat(ms->pipe.data_ptr,data,*data_size);
624 /* TBD: check return value */
625 *data_size = 0;
626 }
627 else {
628 if (!ms->pipe.data_ptr) {
629 return MHD_NO;
630 }
631 if (is_reserved(ms->attr,reserved_attr)) {
632 resp = MHD_create_response_from_data(
633 0,NULL,MHD_NO,MHD_NO);
634 if (!resp) {
635 return MHD_NO;
636 }
637 MHD_queue_response(conn,MHD_HTTP_BAD_REQUEST,
638 resp);
639 MHD_destroy_response(resp);
640 free_ms(ms);
641 return MHD_YES;
642 }
643 meta_set_value(ms->bucket,ms->key,ms->attr,ms->pipe.data_ptr);
644 /*
645 * We should always re-replicate, because the replication
646 * policy might refer to this attr.
647 */
648 DPRINTF("rereplicate (attr PUT)\n");
649 recheck_replication(ms,NULL);
650 free_ms(ms);
651 send_resp = 1;
652 }
653
654 if (send_resp) {
655 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
656 if (!resp) {
657 return MHD_NO;
658 }
659 MHD_queue_response(conn,MHD_HTTP_CREATED,resp);
660 MHD_destroy_response(resp);
661 /*
662 * TBD: check if the attribute was a replication policy, and
663 * start/stop replication activities as appropriate.
664 */
665 }
666
667 return MHD_YES;
668 }
669
670 /**********
671 * For queries, we have to deal with MHD's post-iterator interface (not
672 * quite the same as the data-iteration even though we use it that way) on
673 * one side, and a query-iterator interface on the other. Data on both
674 * sides could be quite large, so we can't just stick them in header lines.
675 * We do still buffer the query in memory, though. Once that's done, we do
676 * very simple parsing - it will be more complicated later - and create the
677 * query iterator. That's also driven by MHD, this time though the
678 * content-callback interface, and repeatedly calls in to the metadata
679 * module to fetch one object name at a time.
680 **********/
681
682 int
683 query_iterator (void *ctx, enum MHD_ValueKind kind, const char *key,
684 const char *filename, const char *content_type,
685 const char *transfer_encoding, const char *data,
686 uint64_t off, size_t size)
687 {
688 (void)ctx;
689 (void)kind;
690 (void)key;
691 (void)filename;
692 (void)content_type;
693 (void)transfer_encoding;
694 (void)data;
695 (void)off;
696 (void)size;
697
698 /* We actually accumulate the data in proxy_query. */
699 return MHD_YES;
700 }
701
702 /* MHD reader function during queries. Return -1 for EOF. */
703 int
704 proxy_query_func (void *ctx, uint64_t pos, char *buf, int max)
705 {
706 my_state *ms = ctx;
707 int len;
708 const char *accept_hdr;
709 char *bucket;
710 char *key;
711
712 (void)pos;
713
714 accept_hdr = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,
715 "Accept");
716
717 if (!ms->gen_ctx) {
718 ms->gen_ctx = tmpl_get_ctx(accept_hdr);
719 if (!ms->gen_ctx) {
720 return -1;
721 }
722 ms->cleanup |= CLEANUP_TMPL;
723 len = tmpl_obj_header(ms->gen_ctx);
724 if (!len) {
725 return -1;
726 }
727 if (len > max) {
728 len = max;
729 }
730 memcpy(buf,ms->gen_ctx->buf,len);
731 return len;
732 }
733
734 if (ms->gen_ctx == TMPL_CTX_DONE) {
735 return -1;
736 }
737
738 for(;;) {
739 if (!meta_query_next(ms->query,&bucket,&key)) {
740 break;
741 }
742 if (is_reserved(key,reserved_name)) {
743 continue;
744 }
745 len = tmpl_obj_entry(ms->gen_ctx,bucket,key);
746 if (!len) {
747 return -1;
748 }
749 if (len > max) {
750 len = max;
751 }
752 memcpy(buf,ms->gen_ctx->buf,len);
753 return len;
754 }
755
756 len = tmpl_obj_footer(ms->gen_ctx);
757 if (!len) {
758 return -1;
759 }
760 if (len > max) {
761 len = max;
762 }
763 memcpy(buf,ms->gen_ctx->buf,len);
764 free(ms->gen_ctx);
765 ms->cleanup &= ~CLEANUP_TMPL;
766 ms->gen_ctx = TMPL_CTX_DONE;
767 return len;
768 }
769
770 int
771 proxy_query (void *cctx, struct MHD_Connection *conn, const char *url,
772 const char *method, const char *version, const char *data,
773 size_t *data_size, void **rctx)
774 {
775 struct MHD_Response *resp;
776 my_state *ms = *rctx;
777
778 (void)cctx;
779 (void)method;
780 (void)version;
781
782 DPRINTF("PROXY QUERY %s (%zu)\n",url,*data_size);
783
784 if (ms->state == MS_NEW) {
785 ms->state = MS_NORMAL;
786 ms->post = MHD_create_post_processor(conn,4096,
787 query_iterator,ms);
788 ms->cleanup |= CLEANUP_POST;
789 }
790 else if (*data_size) {
791 MHD_post_process(ms->post,data,*data_size);
792 if (ms->pipe.data_len) {
793 ms->pipe.data_len += *data_size;
794 char *p = realloc(ms->pipe.data_ptr,ms->pipe.data_len);
795 if (!p) {
796 return MHD_NO;
797 }
798 ms->pipe.data_ptr = p;
799 }
800 else {
801 ms->pipe.data_len = *data_size + 1;
802 ms->pipe.data_ptr = malloc(ms->pipe.data_len);
803 if (!ms->pipe.data_ptr) {
804 return MHD_NO;
805 }
806 ((char *)ms->pipe.data_ptr)[0] = '\0';
807 ms->cleanup |= CLEANUP_BUF_PTR;
808 }
809 (void)strncat(ms->pipe.data_ptr,data,*data_size);
810 /* TBD: check return value */
811 *data_size = 0;
812 }
813 else {
814 if (!ms->pipe.data_ptr) {
815 return MHD_NO;
816 }
817 ms->query = meta_query_new(ms->bucket,NULL,ms->pipe.data_ptr);
818 ms->cleanup |= CLEANUP_QUERY;
819 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
820 65536, proxy_query_func, ms, simple_closer);
821 if (!resp) {
822 fprintf(stderr,"MHD_crfc failed\n");
823 simple_closer(ms);
824 return MHD_NO;
825 }
826 MHD_add_response_header(resp,"Content-Type","text/xml");
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 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 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,(char *)url);
888 if (rc != MHD_YES) {
889 return rc;
890 }
891
892 copied_url = strdup(url);
893 assert (copied_url);
894 bucket = strtok_r(copied_url,"/",&stctx);
895 key = strtok_r(NULL,"/",&stctx);
896 meta_delete(bucket,key);
897 free(copied_url);
898
899 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
900 if (!resp) {
901 return MHD_NO;
902 }
903 MHD_add_response_header(resp,"Content-Type","text/xml");
904 MHD_queue_response(conn,MHD_HTTP_OK,resp);
905 MHD_destroy_response(resp);
906
907 replicate_delete((char *)url);
908 free_ms(ms);
909 return MHD_YES;
910 }
911
912 /* TBD: get actual bucket list */
913 typedef struct {
914 char *rel;
915 char *link;
916 } fake_bucket_t;
917
918 fake_bucket_t fake_bucket_list[] = {
919 { "bucket_factory", "_new" },
920 { "provider_list", "_providers" },
921 };
922
923 int
924 root_blob_generator (void *ctx, uint64_t pos, char *buf, int max)
925 {
926 my_state *ms = ctx;
927 fake_bucket_t * fb;
928 int len;
929 const char *accept_hdr;
930 const char *host;
931 char *bucket;
932 char *key;
933
934 (void)pos;
935
936 accept_hdr = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,
937 "Accept");
938 host = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,"Host");
939
940 if (!ms->gen_ctx) {
941 ms->gen_ctx = tmpl_get_ctx(accept_hdr);
942 if (!ms->gen_ctx) {
943 return -1;
944 }
945 ms->cleanup |= CLEANUP_TMPL;
946 ms->gen_ctx->base = host;
947 len = tmpl_root_header(ms->gen_ctx,"image_warehouse","1.0");
948 if (!len) {
949 return -1;
950 }
951 if (len > max) {
952 len = max;
953 }
954 memcpy(buf,ms->gen_ctx->buf,len);
955 return len;
956 }
957
958 if (ms->gen_ctx == TMPL_CTX_DONE) {
959 return -1;
960 }
961
962 if (ms->gen_ctx->index < ARRAY_SIZE(fake_bucket_list)) {
963 fb = fake_bucket_list + ms->gen_ctx->index;
964 len = tmpl_root_entry(ms->gen_ctx,fb->rel,fb->link);
965 if (!len) {
966 return -1;
967 }
968 if (len > max) {
969 len = max;
970 }
971 memcpy(buf,ms->gen_ctx->buf,len);
972 return len;
973 }
974
975 if (meta_query_next(ms->query,&bucket,&key)) {
976 len = tmpl_root_entry(ms->gen_ctx,"bucket",bucket);
977 if (!len) {
978 return -1;
979 }
980 if (len > max) {
981 len = max;
982 }
983 memcpy(buf,ms->gen_ctx->buf,len);
984 return len;
985 }
986
987 len = tmpl_root_footer(ms->gen_ctx);
988 if (!len) {
989 return -1;
990 }
991 if (len > max) {
992 len = max;
993 }
994 memcpy(buf,ms->gen_ctx->buf,len);
995 free(ms->gen_ctx);
996 ms->cleanup &= ~CLEANUP_TMPL;
997 ms->gen_ctx = TMPL_CTX_DONE;
998 return len;
999 }
1000
1001 int
1002 proxy_api_root (void *cctx, struct MHD_Connection *conn, const char *url,
1003 const char *method, const char *version, const char *data,
1004 size_t *data_size, void **rctx)
1005 {
1006 struct MHD_Response *resp = NULL;
1007 unsigned int rc = MHD_HTTP_OK;
1008 my_state *ms = *rctx;
1009
1010 (void)cctx;
1011 (void)method;
1012 (void)version;
1013 (void)data;
1014
1015 DPRINTF("PROXY API ROOT (%s, %zu)\n",url,*data_size);
1016
1017 ms->query = meta_query_new(NULL,"_default",NULL);
1018 if (!ms->query) {
1019 free_ms(ms);
1020 return MHD_NO;
1021 }
1022 ms->cleanup |= CLEANUP_QUERY;
1023
1024 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
1025 65536, root_blob_generator, ms, simple_closer);
1026 if (!resp) {
1027 return MHD_NO;
1028 }
1029 MHD_add_response_header(resp,"Content-Type","text/xml");
1030 MHD_queue_response(conn,rc,resp);
1031 MHD_destroy_response(resp);
1032
1033 return MHD_YES;
1034
1035 }
1036
1037 int
1038 post_iterator (void *ctx, enum MHD_ValueKind kind, const char *key,
1039 const char *filename, const char *content_type,
1040 const char *transfer_encoding, const char *data,
1041 uint64_t off, size_t size)
1042 {
1043 char *old_val;
1044 size_t old_len;
1045 char *new_val;
1046
1047 (void)kind;
1048 (void)filename;
1049 (void)content_type;
1050 (void)transfer_encoding;
1051 (void)off;
1052
1053 printf("adding %s, size=%zu\n",key,size);
1054
1055 // TBD: don't assume that values are null-terminated strings
1056 old_val = g_hash_table_lookup(ctx,key);
1057 if (old_val) {
1058 old_len = strlen(old_val);
1059 new_val = malloc(old_len+size+1);
1060 if (!new_val) {
1061 return MHD_NO;
1062 }
1063 memcpy(new_val,old_val,old_len);
1064 memcpy(new_val+old_len,data,size);
1065 new_val[old_len+size] = '\0';
1066 }
1067 else {
1068 new_val = malloc(size+1);
1069 if (!new_val) {
1070 return MHD_NO;
1071 }
1072 memcpy(new_val,data,size);
1073 new_val[size] = '\0';
1074 }
1075
1076 g_hash_table_insert(ctx,strdup(key),new_val);
1077 /* TBD: check return value for strdups (none avail for insert) */
1078 return MHD_YES;
1079 }
1080
1081 /* Returns TRUE if we found an *invalid* key. */
1082 gboolean
1083 post_find (gpointer key, gpointer value, gpointer ctx)
1084 {
1085 (void)value;
1086 (void)ctx;
1087
1088 if (!is_reserved(key,reserved_attr)) {
1089 return FALSE;
1090 }
1091
1092 DPRINTF("bad attr %s\n", (char *)key);
1093 return TRUE;
1094 }
1095
1096 void
1097 post_foreach (gpointer key, gpointer value, gpointer ctx)
1098 {
1099 my_state *ms = ctx;
1100
1101 DPRINTF("setting %s = %s for %s/%s\n",(char *)key, (char *)value,
1102 ms->bucket,ms->key);
1103 meta_set_value(ms->bucket,ms->key,key,value);
1104 }
1105
1106 int
1107 create_bucket (char *name)
1108 {
1109 int rc;
1110
1111 if (is_reserved(name,reserved_name)) {
1112 return MHD_HTTP_BAD_REQUEST;
1113 }
1114
1115 rc = main_func_tbl->bcreate_func(name);
1116 if (rc == MHD_HTTP_OK) {
1117 if (meta_set_value(name,"_default", "_policy","0") != 0) {
1118 DPRINTF("default-policy " "create failed\n");
1119 /* Non-fatal. */
1120 }
1121 DPRINTF("created bucket %s\n",name);
1122 /*
1123 * There's not a whole lot to do about bucket-creation
1124 * failures on replicas, other than to report them, unless
1125 * we adopt an "all or nothing" approach and unwind the
1126 * create on the primary as well. Then what if that fails?
1127 * It's just one example of the general "fewer replicas
1128 * than desired" distributed-system problem, not worth a
1129 * point solution here/now. Revisit when we have a more
1130 * general replica-repair policy/system in place.
1131 */
1132 replicate_bcreate(name);
1133 }
1134
1135 return rc;
1136 }
1137
1138 int
1139 proxy_bucket_post (void *cctx, struct MHD_Connection *conn, const char *url,
1140 const char *method, const char *version, const char *data,
1141 size_t *data_size, void **rctx)
1142 {
1143 struct MHD_Response *resp;
1144 my_state *ms = *rctx;
1145 int rc;
1146 char *key;
1147
1148 (void)cctx;
1149 (void)method;
1150 (void)version;
1151
1152 DPRINTF("PROXY POST (%s, %zu)\n",url,*data_size);
1153
1154 if (ms->state == MS_NEW) {
1155 ms->state = MS_NORMAL;
1156 ms->url = (char *)url;
1157 ms->dict = g_hash_table_new_full(
1158 g_str_hash,g_str_equal,free,free);
1159 ms->cleanup |= CLEANUP_DICT;
1160 ms->post = MHD_create_post_processor(conn,4096,
1161 post_iterator,ms->dict);
1162 ms->cleanup |= CLEANUP_POST;
1163 }
1164 else if (*data_size) {
1165 MHD_post_process(ms->post,data,*data_size);
1166 *data_size = 0;
1167 }
1168 else {
1169 rc = MHD_HTTP_BAD_REQUEST;
1170 key = g_hash_table_lookup(ms->dict,"key");
1171 if (key) {
1172 strncpy(ms->key,key,MAX_FIELD_LEN-1);
1173 g_hash_table_remove(ms->dict,"key");
1174 if (!g_hash_table_find(ms->dict,post_find,ms)) {
1175 g_hash_table_foreach(ms->dict,post_foreach,ms);
1176 DPRINTF("rereplicate (bucket POST)\n");
1177 recheck_replication(ms,NULL);
1178 rc = MHD_HTTP_OK;
1179 }
1180 }
1181 else if (!strcmp(ms->bucket,"_new")) {
1182 key = g_hash_table_lookup(ms->dict,"name");
1183 if (key != NULL) {
1184 rc = create_bucket(key);
1185 }
1186 }
1187 else {
1188 DPRINTF("A parameter is MISSING (fail)\n");
1189 }
1190 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1191 if (!resp) {
1192 fprintf(stderr,"MHD_crfd failed\n");
1193 return MHD_NO;
1194 }
1195 MHD_queue_response(conn,rc,resp);
1196 MHD_destroy_response(resp);
1197 free_ms(ms);
1198 }
1199
1200 return MHD_YES;
1201 }
1202
1203 int
1204 check_location (my_state *ms)
1205 {
1206 char *loc = g_hash_table_lookup(ms->dict,"depot");
1207
1208 if (!loc) {
1209 DPRINTF("missing loc on check for %s/%s\n",ms->bucket,ms->key);
1210 return MHD_HTTP_BAD_REQUEST;
1211 }
1212
1213 if (!meta_has_copy(ms->bucket,ms->key,loc)) {
1214 DPRINTF("did not find %s/%s at %s\n",ms->bucket,ms->key,loc);
1215 return MHD_HTTP_NOT_FOUND;
1216 }
1217
1218 /* TBD: meta_has_copy returns an etag which we should check */
1219 DPRINTF("found %s/%s at %s\n",ms->bucket,ms->key,loc);
1220 return MHD_HTTP_OK;
1221 }
1222
1223 int
1224 register_image (my_state *ms)
1225 {
1226 char *site;
1227 provider_t prov;
1228 int i;
1229 char *next;
1230
1231 site = g_hash_table_lookup(ms->dict,"site");
1232 if (!site) {
1233 printf("site MISSING\n");
1234 return MHD_HTTP_BAD_REQUEST;
1235 }
1236
1237 next = index(site,':');
1238 if (next) {
1239 *(next++) = '\0';
1240 }
1241
1242 for (i = 0; get_provider(i,&prov); ++i) {
1243 if (strcmp(prov.name,site)) {
1244 continue;
1245 }
1246 return prov.func_tbl->register_func(ms,&prov,next,ms->dict);
1247 }
1248
1249 DPRINTF("site %s not found\n",site);
1250 return MHD_HTTP_BAD_REQUEST;
1251 }
1252
1253 int
1254 proxy_object_post (void *cctx, struct MHD_Connection *conn, const char *url,
1255 const char *method, const char *version, const char *data,
1256 size_t *data_size, void **rctx)
1257 {
1258 struct MHD_Response *resp;
1259 my_state *ms = *rctx;
1260 int rc;
1261 char *op;
1262
1263 (void)cctx;
1264 (void)method;
1265 (void)version;
1266
1267 DPRINTF("PROXY POST (%s, %zu)\n",url,*data_size);
1268
1269 if (ms->state == MS_NEW) {
1270 ms->state = MS_NORMAL;
1271 ms->url = (char *)url;
1272 ms->dict = g_hash_table_new_full(
1273 g_str_hash,g_str_equal,free,free);
1274 ms->cleanup |= CLEANUP_DICT;
1275 ms->post = MHD_create_post_processor(conn,4096,
1276 post_iterator,ms->dict);
1277 ms->cleanup |= CLEANUP_POST;
1278 }
1279 else if (*data_size) {
1280 MHD_post_process(ms->post,data,*data_size);
1281 *data_size = 0;
1282 }
1283 else {
1284 rc = MHD_HTTP_BAD_REQUEST;
1285 if (!g_hash_table_find(ms->dict,post_find,ms)) {
1286 op = g_hash_table_lookup(ms->dict,"op");
1287 if (op) {
1288 if (!strcmp(op,"push")) {
1289 DPRINTF("rereplicate (obj POST)\n");
1290 recheck_replication(ms,NULL);
1291 rc = MHD_HTTP_OK;
1292 }
1293 else if (!strcmp(op,"check")) {
1294 rc = check_location(ms);
1295 }
1296 else if (!strcmp(op,"register")) {
1297 rc = register_image(ms);
1298 }
1299 else {
1300 DPRINTF("unknown op %s for %s/%s\n",
1301 op, ms->bucket, ms->key);
1302 }
1303 }
1304 else {
1305 DPRINTF("op is MISSING (fail)\n");
1306 }
1307 }
1308 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1309 if (!resp) {
1310 fprintf(stderr,"MHD_crfd failed\n");
1311 return MHD_NO;
1312 }
1313 MHD_queue_response(conn,rc,resp);
1314 MHD_destroy_response(resp);
1315 free_ms(ms);
1316 }
1317
1318 return MHD_YES;
1319
1320 }
1321
1322
1323 int
1324 prov_list_generator (void *ctx, uint64_t pos, char *buf, int max)
1325 {
1326 my_state *ms = ctx;
1327 int len;
1328 provider_t prov;
1329 const char *accept_hdr;
1330
1331 (void)pos;
1332
1333 accept_hdr = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,
1334 "Accept");
1335
1336 if (!ms->gen_ctx) {
1337 ms->gen_ctx = tmpl_get_ctx(accept_hdr);
1338 if (!ms->gen_ctx) {
1339 return -1;
1340 }
1341 ms->cleanup |= CLEANUP_TMPL;
1342 len = tmpl_prov_header(ms->gen_ctx);
1343 if (!len) {
1344 return -1;
1345 }
1346 if (len > max) {
1347 len = max;
1348 }
1349 memcpy(buf,ms->gen_ctx->buf,len);
1350 return len;
1351 }
1352
1353 if (ms->gen_ctx == TMPL_CTX_DONE) {
1354 return -1;
1355 }
1356
1357 if (get_provider(ms->gen_ctx->index,&prov)) {
1358 len = tmpl_prov_entry(ms->gen_ctx,prov.name,prov.type,
1359 prov.host, prov.port, prov.username, prov.password);
1360 if (!len) {
1361 return -1;
1362 }
1363 if (len > max) {
1364 len = max;
1365 }
1366 memcpy(buf,ms->gen_ctx->buf,len);
1367 return len;
1368 }
1369
1370 len = tmpl_prov_footer(ms->gen_ctx);
1371 if (!len) {
1372 return -1;
1373 }
1374 if (len > max) {
1375 len = max;
1376 }
1377 memcpy(buf,ms->gen_ctx->buf,len);
1378 free(ms->gen_ctx);
1379 ms->cleanup &= ~CLEANUP_TMPL;
1380 ms->gen_ctx = TMPL_CTX_DONE;
1381 return len;
1382 }
1383
1384 int
1385 proxy_list_provs (void *cctx, struct MHD_Connection *conn, const char *url,
1386 const char *method, const char *version, const char *data,
1387 size_t *data_size, void **rctx)
1388 {
1389 struct MHD_Response *resp;
1390 my_state *ms = *rctx;
1391
1392 (void)cctx;
1393 (void)url;
1394 (void)method;
1395 (void)version;
1396 (void)data;
1397 (void)data_size;
1398
1399 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
1400 65536, prov_list_generator, ms, simple_closer);
1401 if (!resp) {
1402 fprintf(stderr,"MHD_crfd failed\n");
1403 simple_closer(ms);
1404 return MHD_NO;
1405 }
1406 MHD_queue_response(conn,MHD_HTTP_OK,resp);
1407 MHD_destroy_response(resp);
1408
1409 return MHD_YES;
1410 }
1411
1412 int
1413 prov_iterator (void *ctx, enum MHD_ValueKind kind, const char *key,
1414 const char *filename, const char *content_type,
1415 const char *transfer_encoding, const char *data,
1416 uint64_t off, size_t size)
1417 {
1418 (void)kind;
1419 (void)filename;
1420 (void)content_type;
1421 (void)transfer_encoding;
1422 (void)off;
1423
1424 g_hash_table_insert(ctx,strdup(key),strndup(data,size));
1425 /* TBD: check return value for strdups (none avail for insert) */
1426 return MHD_YES;
1427 }
1428
1429
1430 int
1431 proxy_update_prov (void *cctx, struct MHD_Connection *conn, const char *url,
1432 const char *method, const char *version, const char *data,
1433 size_t *data_size, void **rctx)
1434 {
1435 struct MHD_Response *resp;
1436 my_state *ms = *rctx;
1437 int rc;
1438 char *provider;
1439 char *username;
1440 char *password;
1441
1442 (void)cctx;
1443 (void)method;
1444 (void)version;
1445
1446 if (ms->state == MS_NEW) {
1447 ms->state = MS_NORMAL;
1448 ms->url = (char *)url;
1449 ms->dict = g_hash_table_new_full(
1450 g_str_hash,g_str_equal,free,free);
1451 ms->cleanup |= CLEANUP_DICT;
1452 ms->post = MHD_create_post_processor(conn,4096,
1453 prov_iterator,ms->dict);
1454 ms->cleanup |= CLEANUP_POST;
1455 }
1456 else if (*data_size) {
1457 MHD_post_process(ms->post,data,*data_size);
1458 *data_size = 0;
1459 }
1460 else {
1461 rc = MHD_HTTP_BAD_REQUEST;
1462 provider = g_hash_table_lookup(ms->dict,"provider");
1463 username = g_hash_table_lookup(ms->dict,"username");
1464 password = g_hash_table_lookup(ms->dict,"password");
1465 if (provider && username && password) {
1466 update_provider(provider,username,password);
1467 rc = MHD_HTTP_OK;
1468 }
1469 else {
1470 DPRINTF("provider/username/password MISSING\n");
1471 }
1472 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1473 if (!resp) {
1474 fprintf(stderr,"MHD_crfd failed\n");
1475 return MHD_NO;
1476 }
1477 MHD_queue_response(conn,rc,resp);
1478 MHD_destroy_response(resp);
1479 free_ms(ms);
1480 }
1481
1482 return MHD_YES;
1483 }
1484
1485 int
1486 proxy_create_bucket (void *cctx, struct MHD_Connection *conn, const char *url,
1487 const char *method, const char *version, const char *data,
1488 size_t *data_size, void **rctx)
1489 {
1490 struct MHD_Response *resp;
1491 my_state *ms = *rctx;
1492 int rc;
1493
1494 (void)cctx;
1495 (void)method;
1496 (void)version;
1497 (void)data;
1498 (void)data_size;
1499 (void)url;
1500
1501 /* curl -T moo.empty http://localhost:9090/_new by accident */
1502 rc = create_bucket(ms->bucket);
1503
1504 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1505 if (!resp) {
1506 fprintf(stderr,"MHD_crfd failed\n");
1507 return MHD_NO;
1508 }
1509 MHD_queue_response(conn,rc,resp);
1510 MHD_destroy_response(resp);
1511
1512 free_ms(ms);
1513 return MHD_YES;
1514 }
1515
1516 rule my_rules[] = {
1517 { /* get bucket list */
1518 "GET", URL_ROOT, proxy_api_root },
1519 { /* get object list */
1520 "GET", URL_BUCKET, proxy_list_objs },
1521 { /* create bucket */
1522 "PUT", URL_BUCKET, proxy_create_bucket },
1523 { /* get object data */
1524 "GET", URL_OBJECT, proxy_get_data },
1525 { /* get attribute data */
1526 "GET", URL_ATTR, proxy_get_attr },
1527 { /* put object data */
1528 "PUT", URL_OBJECT, proxy_put_data },
1529 { /* put attribute data */
1530 "PUT", URL_ATTR, proxy_put_attr },
1531 { /* create object and/or modify attributes */
1532 "POST", URL_BUCKET, proxy_bucket_post },
1533 { /* perform control operations on an object */
1534 "POST", URL_OBJECT, proxy_object_post },
1535 { /* query */
1536 "POST", URL_QUERY, proxy_query },
1537 { /* delete object */
1538 "DELETE", URL_OBJECT, proxy_delete },
1539 { /* delete attribute (TBD) */
1540 "DELETE", URL_ATTR, NULL },
1541 { /* get provider list */
1542 "GET", URL_PROVLIST, proxy_list_provs },
1543 { /* update a provider */
1544 "POST", URL_PROVLIST, proxy_update_prov },
1545 { NULL, 0, NULL }
1546 };
1547
1548 url_type
1549 parse_url (const char *url, my_state *ms)
1550 {
1551 unsigned short esize;
1552 unsigned short eindex;
1553 char *parts[URL_INVAL];
1554
1555 if (strstr(url,"../")) {
1556 /* Naughty, naughty. Never a good reason to allow this. */
1557 DPRINTF("Rejecting ../ in path.\n");
1558 return URL_INVAL;
1559 }
1560
1561 eindex = URL_ROOT;
1562 parts[URL_BUCKET] = ms->bucket;
1563 parts[URL_OBJECT] = ms->key;
1564 parts[URL_ATTR] = ms->attr;
1565
1566 for (;;) {
1567 while (*url == '/') {
1568 ++url;
1569 }
1570
1571 if (!*url) {
1572 if (eindex == URL_BUCKET) {
1573 if (!strcmp(ms->bucket,"_providers")) {
1574 eindex = URL_PROVLIST;
1575 }
1576 }
1577 else if (eindex == URL_OBJECT) {
1578 if (!strcmp(ms->key,"_query")) {
1579 eindex = URL_QUERY;
1580 }
1581 }
1582 break;
1583 }
1584
1585 if (++eindex >= URL_INVAL) {
1586 return URL_INVAL;
1587 }
1588 esize = 0;
1589
1590 while (*url && (*url != '/')) {
1591 parts[eindex][esize++] = *(url++);
1592 if (esize >= MAX_FIELD_LEN) {
1593 return URL_INVAL;
1594 }
1595 }
1596 }
1597
1598 return eindex;
1599 }
1600
1601 int
1602 access_handler (void *cctx, struct MHD_Connection *conn, const char *url,
1603 const char *method, const char *version, const char *data,
1604 size_t *data_size, void **rctx)
1605 {
1606 unsigned int i;
1607 url_type utype;
1608 struct MHD_Response *resp;
1609 my_state *ms = *rctx;
1610
1611 if (ms) {
1612 return ms->handler(cctx,conn,url,method,version,
1613 data,data_size,rctx);
1614 }
1615
1616 ms = calloc(sizeof(*ms), 1);
1617 if (!ms) {
1618 return MHD_NO;
1619 }
1620
1621 utype = parse_url(url,ms);
1622
1623 for (i = 0; my_rules[i].method; ++i) {
1624 if (utype != my_rules[i].utype) {
1625 continue;
1626 }
1627 if (strcmp(method,my_rules[i].method)) {
1628 continue;
1629 }
1630 if (!my_rules[i].handler) {
1631 break;
1632 }
1633 ms->handler = my_rules[i].handler;
1634 ms->state = MS_NEW;
1635 ms->fd = (-1);
1636 ms->url = NULL;
1637 ms->post = NULL;
1638 ms->conn = conn;
1639 *rctx = ms;
1640 return ms->handler(cctx,conn,url,method,version,
1641 data,data_size,rctx);
1642 }
1643
1644 if (!strcmp(method,"QUIT")) {
1645 (void)sem_post((sem_t *)cctx);
1646 return MHD_NO;
1647 }
1648
1649 fprintf(stderr,"bad request m=%s u=%s\n",method,url);
1650 free_ms(ms);
1651
1652 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1653 if (!resp) {
1654 return MHD_NO;
1655 }
1656 MHD_queue_response(conn,MHD_HTTP_NOT_FOUND,resp);
1657 MHD_destroy_response(resp);
1658 return MHD_YES;
1659 }
1660
1661 /* These enum values cannot possibly conflict with the option values
1662 ordinarily used by commands, including CHAR_MAX + 1, etc. Avoid
1663 CHAR_MIN - 1, as it may equal -1, the getopt end-of-options value. */
1664 enum
1665 {
1666 GETOPT_HELP_CHAR = (CHAR_MIN - 2),
1667 GETOPT_VERSION_CHAR = (CHAR_MIN - 3)
1668 };
1669
1670 struct option my_options[] = {
1671 { "config", required_argument, NULL, 'c' },
1672 { "db", required_argument, NULL, 'd' },
1673 { "master", required_argument, NULL, 'm' },
1674 { "port", required_argument, NULL, 'p' },
1675 { "verbose", no_argument, NULL, 'v' },
1676 { "version", no_argument, NULL, GETOPT_VERSION_CHAR },
1677 { "help", no_argument, NULL, GETOPT_HELP_CHAR },
1678 { NULL, 0, NULL, '\0' }
1679 };
1680
1681 void
1682 usage (int status)
1683 {
1684 if (status != EXIT_SUCCESS)
1685 fprintf (stderr, _("Try `%s --help' for more information.\n"),
1686 program_name);
1687 else
1688 {
1689 printf (_("\
1690 Usage: %s [OPTION]\n\
1691 "),
1692 program_name);
1693 fputs (_("\
1694 Deltacloud image-warehouse daemon.\n\
1695 A configuration file must be specified.\n\
1696 \n\
1697 -c, --config=FILE config file [required]\n\
1698 -d, --db=HOST_PORT database server as ip[:port]\n\
1699 -m, --master=HOST_PORT master (upstream) server as ip[:port]\n\
1700 -p, --port=PORT alternate listen port (default 9090)\n\
1701 -v, --verbose verbose/debug output\n\
1702 \n\
1703 --help display this help and exit\n\
1704 --version output version information and exit\n\
1705 "), stdout);
1706 printf (_("\
1707 \n\
1708 Report %s bugs to %s.\n\
1709 "),
1710 program_name, PACKAGE_BUGREPORT);
1711 }
1712 exit (status);
1713 }
1714
1715 int
1716 main (int argc, char **argv)
1717 {
1718 struct MHD_Daemon *the_daemon;
1719 sem_t the_sem;
1720 char *stctx = NULL;
1721 char *port_tmp;
1722
1723 program_name = argv[0];
1724
1725 for (;;) switch (getopt_long(argc,argv,"c:d:m:p:v",my_options,NULL)) {
1726 case 'c':
1727 cfg_file = optarg;
1728 break;
1729 case 'd':
1730 assert (optarg);
1731 db_host = strtok_r(optarg,":",&stctx);
1732 port_tmp = strtok_r(NULL,":",&stctx);
1733 if (port_tmp) {
1734 db_port = (unsigned short)strtoul(port_tmp,NULL,10);
1735 }
1736 break;
1737 case 'm':
1738 assert (optarg);
1739 master_host = strtok_r(optarg,":",&stctx);
1740 port_tmp = strtok_r(NULL,":",&stctx);
1741 if (port_tmp) {
1742 master_port = (unsigned short)strtoul(port_tmp,NULL,10);
1743 }
1744 break;
1745 case 'p':
1746 my_port = (unsigned short)strtoul(optarg,NULL,10);
1747 break;
1748 case 'v':
1749 ++verbose;
1750 break;
1751 case GETOPT_HELP_CHAR:
1752 usage(EXIT_SUCCESS);
1753 break;
1754 case GETOPT_VERSION_CHAR:
1755 printf ("%s version %s\n", program_name, PACKAGE_VERSION);
1756 exit (EXIT_SUCCESS);
1757 break;
1758
1759 case -1:
1760 goto args_done;
1761 default:
1762 usage(EXIT_FAILURE);
1763 break;
1764 }
1765 args_done:
1766
1767 if (!cfg_file) {
1768 error (0, 0, "no configuration file specified");
1769 usage (EXIT_FAILURE);
1770 }
1771
1772 me = parse_config();
1773 if (!me) {
1774 fprintf(stderr,"could not parse %s\n",cfg_file);
1775 return !0;
1776 }
1777
1778 sem_init(&the_sem,0,0);
1779 if (proxy_host) {
1780 if (s3mode) {
1781 main_func_tbl = &s3_func_tbl;
1782 }
1783 else {
1784 main_func_tbl = &curl_func_tbl;
1785 }
1786 }
1787 else {
1788 main_func_tbl = &fs_func_tbl;
1789 }
1790
1791 if (verbose) {
1792 printf("primary store type is %s\n",main_func_tbl->name);
1793 if (master_host) {
1794 printf("operating as slave to %s:%u\n",
1795 master_host, master_port);
1796 }
1797 printf("db is at %s:%u\n",db_host,db_port);
1798 printf("will listen on port %u\n",my_port);
1799 printf("my location is \"%s\"\n",me);
1800 fflush(stdout);
1801 }
1802
1803 backend_init();
1804 main_func_tbl->init_func();
1805 meta_init();
1806 repl_init();
1807
1808 /*
1809 * Gotcha: if we don't set the connection memory limit explicitly,
1810 * the per-connection buffer for MHD will be smaller than that used
1811 * by CURL, so proxy_writefunc will never be able to do its job.
1812 */
1813 the_daemon = MHD_start_daemon(MY_MHD_FLAGS,
1814 my_port, NULL, NULL, &access_handler, &the_sem,
1815 MHD_OPTION_CONNECTION_MEMORY_LIMIT, (size_t)1048576,
1816 MHD_OPTION_END);
1817 if (!the_daemon) {
1818 fprintf(stderr,"Could not create daemon.\n");
1819 return !0;
1820 }
1821
1822 sem_wait(&the_sem);
1823 return 0;
1824 }
image warehouse daemon