bluesky/store-azure.c

   1 /* Blue Sky: File Systems in the Cloud
   2  *
   3  * Copyright (C) 2009  The Regents of the University of California
   4  * Written by Michael Vrable <mvrable@cs.ucsd.edu>
   5  *
   6  * TODO: Licensing
   7  */
   8
   9 #include <stdint.h>
  10 #include <stdlib.h>
  11 #include <glib.h>
  12 #include <string.h>
  13 #include <gcrypt.h>
  14 #include <curl/curl.h>
  15
  16 #include "bluesky-private.h"
  17 #include "libs3.h"
  18
  19 #define AZURE_API_VERSION "2009-09-19"
  20
  21 /* Fixed headers that are used in calculating the request signature for Azure,
  22  * in the order that they are included. */
  23 static const char *signature_headers[] = {
  24     "Content-Encoding", "Content-Language", "Content-Length", "Content-MD5",
  25     "Content-Type", "Date", "If-Modified-Since", "If-Match", "If-None-Match",
  26     "If-Unmodified-Since", "Range", NULL
  27 };
  28
  29 /* Prototype Windows Azure backend for BlueSky.  This is intended to be
  30  * minimally functional, but could use additional work for production use. */
  31
  32 typedef struct {
  33     GThreadPool *thread_pool;
  34     char *account, *container;
  35     uint8_t *key;
  36     size_t key_len;
  37 } AzureStore;
  38
  39 static void get_extra_headers(gchar *key, gchar *value, GList **headers)
  40 {
  41     key = g_ascii_strdown(key, strlen(key));
  42     if (strncmp(key, "x-ms-", strlen("x-ms-")) == 0) {
  43         *headers = g_list_prepend(*headers,
  44                                   g_strdup_printf("%s:%s\n", key, value));
  45     }
  46     g_free(key);
  47 }
  48
  49 static void get_curl_headers(gchar *key, gchar *value,
  50                              struct curl_slist **curl_headers)
  51 {
  52     char *line = g_strdup_printf("%s: %s", key, value);
  53     *curl_headers = curl_slist_append(*curl_headers, line);
  54     g_free(line);
  55 }
  56
  57 struct curlbuf {
  58     /* For reading */
  59     const char *readbuf;
  60     size_t readsize;
  61 };
  62
  63 static size_t curl_readfunc(void *ptr, size_t size, size_t nmemb,
  64                             void *userdata)
  65 {
  66     struct curlbuf *buf = (struct curlbuf *)userdata;
  67
  68     if (buf == NULL)
  69         return 0;
  70
  71     size_t copied = size * nmemb;
  72     if (copied > buf->readsize)
  73         copied = buf->readsize;
  74
  75     memcpy(ptr, buf->readbuf, copied);
  76     buf->readbuf += copied;
  77     buf->readsize -= copied;
  78
  79     return copied;
  80 }
  81
  82 static size_t curl_writefunc(void *ptr, size_t size, size_t nmemb,
  83                              void *userdata)
  84 {
  85     GString *buf = (GString *)userdata;
  86     if (buf != NULL)
  87         g_string_append_len(buf, ptr, size * nmemb);
  88     return size * nmemb;
  89 }
  90
  91 /* Compute the signature for a request to Azure and add it to the headers. */
  92 static void azure_compute_signature(AzureStore *store,
  93                                     GHashTable *headers,
  94                                     const char *method, const char *path)
  95 {
  96     if (g_hash_table_lookup(headers, "Date") == NULL) {
  97         time_t t;
  98         struct tm now;
  99         char timebuf[4096];
 100         time(&t);
 101         gmtime_r(&t, &now);
 102         strftime(timebuf, sizeof(timebuf), "%a, %d %b %Y %H:%M:%S GMT", &now);
 103         g_hash_table_insert(headers, g_strdup("Date"), g_strdup(timebuf));
 104     }
 105
 106     g_hash_table_insert(headers, g_strdup("x-ms-version"),
 107                         g_strdup(AZURE_API_VERSION));
 108
 109     GString *to_sign = g_string_new("");
 110     g_string_append_printf(to_sign, "%s\n", method);
 111     for (const char **h = signature_headers; *h != NULL; h++) {
 112         const char *val = g_hash_table_lookup(headers, *h);
 113         g_string_append_printf(to_sign, "%s\n", val ? val : "");
 114     }
 115
 116     GList *extra_headers = NULL;
 117     g_hash_table_foreach(headers, (GHFunc)get_extra_headers, &extra_headers);
 118     extra_headers = g_list_sort(extra_headers, (GCompareFunc)g_strcmp0);
 119     while (extra_headers != NULL) {
 120         g_string_append(to_sign, extra_headers->data);
 121         g_free(extra_headers->data);
 122         extra_headers = g_list_delete_link(extra_headers, extra_headers);
 123     }
 124
 125     /* FIXME: Doesn't handle query parameters (after '?') */
 126     g_string_append_printf(to_sign, "/%s/%s/%s",
 127                            store->account, store->container, path);
 128
 129     /* Compute an HMAC-SHA-256 of the encoded parameters */
 130     gcry_error_t status;
 131     gcry_md_hd_t handle;
 132     status = gcry_md_open(&handle, GCRY_MD_SHA256, GCRY_MD_FLAG_HMAC);
 133     g_assert(status == 0);
 134     status = gcry_md_setkey(handle, store->key, store->key_len);
 135     g_assert(status == 0);
 136     gcry_md_write(handle, to_sign->str, to_sign->len);
 137     unsigned char *digest = gcry_md_read(handle, GCRY_MD_SHA256);
 138     gchar *signature = g_base64_encode(digest,
 139                                        gcry_md_get_algo_dlen(GCRY_MD_SHA256));
 140     g_hash_table_insert(headers, g_strdup("Authorization"),
 141                         g_strdup_printf("SharedKey %s:%s",
 142                                         store->account, signature));
 143     g_free(signature);
 144     gcry_md_close(handle);
 145     g_string_free(to_sign, TRUE);
 146 }
 147
 148 /* Submit an HTTP request using CURL.  Takes as input the Azure storage backend
 149  * we are acting for, as well as the method (GET, PUT, etc.), HTTP path, other
 150  * HTTP headers, and an optional body.  If body is not NULL, an empty body is
 151  * sent.  This will compute an Azure authentication signature before sending
 152  * the request. */
 153 static BlueSkyRCStr *submit_request(AzureStore *store,
 154                                     const char *method,
 155                                     const char *path,
 156                                     GHashTable *headers,
 157                                     BlueSkyRCStr *body)
 158 {
 159     BlueSkyRCStr *result = NULL;
 160
 161     g_hash_table_insert(headers,
 162                         g_strdup("Content-Length"),
 163                         g_strdup_printf("%zd", body != NULL ? body->len : 0));
 164
 165     if (body != 0 && body->len > 0) {
 166         GChecksum *csum = g_checksum_new(G_CHECKSUM_MD5);
 167         g_checksum_update(csum, (uint8_t *)body->data, body->len);
 168         uint8_t md5[16];
 169         gsize md5_len = sizeof(md5);
 170         g_checksum_get_digest(csum, md5, &md5_len);
 171         g_hash_table_insert(headers,
 172                             g_strdup("Content-MD5"),
 173                             g_base64_encode(md5, md5_len));
 174         g_checksum_free(csum);
 175     }
 176
 177     azure_compute_signature(store, headers, method, path);
 178
 179     CURLcode status;
 180     CURL *curl = curl_easy_init();
 181
 182 #define curl_easy_setopt_safe(opt, val)                                     \
 183     if ((status = curl_easy_setopt(curl, (opt), (val))) != CURLE_OK) {      \
 184         fprintf(stderr, "CURL error: %s!\n", curl_easy_strerror(status));   \
 185         goto cleanup;                                                       \
 186     }
 187
 188     curl_easy_setopt_safe(CURLOPT_NOSIGNAL, 1);
 189     curl_easy_setopt_safe(CURLOPT_NOPROGRESS, 1);
 190     curl_easy_setopt_safe(CURLOPT_NETRC, CURL_NETRC_IGNORED);
 191     curl_easy_setopt_safe(CURLOPT_FOLLOWLOCATION, 1);
 192     curl_easy_setopt_safe(CURLOPT_MAXREDIRS, 10);
 193
 194     curl_easy_setopt_safe(CURLOPT_HEADERFUNCTION, curl_writefunc);
 195     curl_easy_setopt_safe(CURLOPT_HEADERDATA, NULL);
 196
 197     struct curlbuf readbuf;
 198     if (body != NULL) {
 199         readbuf.readbuf = body->data;
 200         readbuf.readsize = body->len;
 201     }
 202     curl_easy_setopt_safe(CURLOPT_READFUNCTION, curl_readfunc);
 203     curl_easy_setopt_safe(CURLOPT_READDATA, body ? &readbuf : NULL);
 204
 205     GString *result_body = g_string_new("");
 206     curl_easy_setopt_safe(CURLOPT_WRITEFUNCTION, curl_writefunc);
 207     curl_easy_setopt_safe(CURLOPT_WRITEDATA, result_body);
 208
 209     struct curl_slist *curl_headers = NULL;
 210     g_hash_table_foreach(headers, (GHFunc)get_curl_headers, &curl_headers);
 211     curl_easy_setopt_safe(CURLOPT_HTTPHEADER, curl_headers);
 212
 213     char *uri = g_strdup_printf("http://%s.blob.core.windows.net/%s/%s",
 214                                 store->account, store->container, path);
 215     printf("URI: %s\n", uri);
 216     curl_easy_setopt_safe(CURLOPT_URL, uri);
 217
 218     if (strcmp(method, "GET") == 0) {
 219         /* nothing special needed */
 220     } else if (strcmp(method, "PUT") == 0) {
 221         curl_easy_setopt_safe(CURLOPT_UPLOAD, 1);
 222     } else if (strcmp(method, "DELETE") == 0) {
 223         curl_easy_setopt_safe(CURLOPT_CUSTOMREQUEST, "DELETE");
 224     }
 225
 226     status = curl_easy_perform(curl);
 227     if (status != 0) {
 228         fprintf(stderr, "CURL error: %s!\n", curl_easy_strerror(status));
 229         goto cleanup;
 230     }
 231
 232     long response_code = 0;
 233     status = curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &response_code);
 234     if (status != 0) {
 235         fprintf(stderr, "CURL error: %s!\n", curl_easy_strerror(status));
 236         goto cleanup;
 237     }
 238
 239     if (response_code / 100 == 2) {
 240         result = bluesky_string_new_from_gstring(result_body);
 241         result_body = NULL;
 242     } else {
 243         fprintf(stderr, "HTTP response code: %ld!\n", response_code);
 244         goto cleanup;
 245     }
 246
 247 cleanup:
 248     if (result != NULL && result_body != NULL)
 249         g_string_free(result_body, TRUE);
 250     curl_easy_cleanup(curl);
 251     curl_slist_free_all(curl_headers);
 252     g_free(uri);
 253
 254     return result;
 255 }
 256
 257 static void azurestore_task(gpointer a, gpointer s)
 258 {
 259     BlueSkyStoreAsync *async = (BlueSkyStoreAsync *)a;
 260     AzureStore *store = (AzureStore *)s;
 261
 262     async->status = ASYNC_RUNNING;
 263     async->exec_time = bluesky_now_hires();
 264
 265     GHashTable *headers = g_hash_table_new_full(g_str_hash, g_str_equal,
 266                                                 g_free, g_free);
 267
 268     BlueSkyRCStr *result = NULL;
 269
 270     if (async->op == STORE_OP_GET) {
 271         result = submit_request(store, "GET", async->key, headers, NULL);
 272         if (result != NULL) {
 273             async->data = result;
 274             async->result = 0;
 275         }
 276     } else if (async->op == STORE_OP_PUT) {
 277         g_hash_table_insert(headers,
 278                             g_strdup("x-ms-blob-type"),
 279                             g_strdup("BlockBlob"));
 280         g_hash_table_insert(headers,
 281                             g_strdup("Transfer-Encoding"),
 282                             g_strdup(""));
 283         result = submit_request(store, "PUT", async->key,
 284                                 headers, async->data);
 285         if (result != NULL) {
 286             async->result = 0;
 287         }
 288         bluesky_string_unref(result);
 289     }
 290
 291     bluesky_store_async_mark_complete(async);
 292     bluesky_store_async_unref(async);
 293     g_hash_table_unref(headers);
 294 }
 295
 296 static gpointer azurestore_new(const gchar *path)
 297 {
 298     AzureStore *store = g_new(AzureStore, 1);
 299     store->thread_pool = g_thread_pool_new(azurestore_task, store, -1, FALSE,
 300                                            NULL);
 301     if (path == NULL || strlen(path) == 0)
 302         store->container = g_strdup("bluesky");
 303     else
 304         store->container = g_strdup(path);
 305
 306     store->account = g_strdup(getenv("AZURE_ACCOUNT_NAME"));
 307
 308     const char *key = getenv("AZURE_SECRET_KEY");
 309     store->key = g_base64_decode(key, &store->key_len);
 310
 311     g_print("Initializing Azure with account %s, container %s\n",
 312             store->account, store->container);
 313
 314     return store;
 315 }
 316
 317 static void azurestore_destroy(gpointer store)
 318 {
 319     /* TODO: Clean up resources */
 320 }
 321
 322 static void azurestore_submit(gpointer s, BlueSkyStoreAsync *async)
 323 {
 324     AzureStore *store = (AzureStore *)s;
 325     g_return_if_fail(async->status == ASYNC_NEW);
 326     g_return_if_fail(async->op != STORE_OP_NONE);
 327
 328     switch (async->op) {
 329     case STORE_OP_GET:
 330     case STORE_OP_PUT:
 331         async->status = ASYNC_PENDING;
 332         bluesky_store_async_ref(async);
 333         g_thread_pool_push(store->thread_pool, async, NULL);
 334         break;
 335
 336     default:
 337         g_warning("Uknown operation type for AzureStore: %d\n", async->op);
 338         bluesky_store_async_mark_complete(async);
 339         break;
 340     }
 341 }
 342
 343 static void azurestore_cleanup(gpointer store, BlueSkyStoreAsync *async)
 344 {
 345 }
 346
 347 static char *azurestore_lookup_last(gpointer s, const char *prefix)
 348 {
 349     return NULL;
 350 }
 351
 352 static BlueSkyStoreImplementation store_impl = {
 353     .create = azurestore_new,
 354     .destroy = azurestore_destroy,
 355     .submit = azurestore_submit,
 356     .cleanup = azurestore_cleanup,
 357     .lookup_last = azurestore_lookup_last,
 358 };
 359
 360 void bluesky_store_init_azure(void)
 361 {
 362     bluesky_store_register(&store_impl, "azure");
 363 }
 364
 365 #if 0
 366 typedef struct {
 367     enum { S3_GET, S3_PUT } op;
 368     gchar *key;
 369     BlueSkyRCStr *data;
 370 } S3Op;
 371
 372 struct get_info {
 373     int success;
 374     GString *buf;
 375 };
 376
 377 struct put_info {
 378     int success;
 379     BlueSkyRCStr *val;
 380     gint offset;
 381 };
 382
 383 struct list_info {
 384     int success;
 385     char *last_entry;
 386     gboolean truncated;
 387 };
 388
 389 static S3Status s3store_get_handler(int bufferSize, const char *buffer,
 390                                     void *callbackData)
 391 {
 392     struct get_info *info = (struct get_info *)callbackData;
 393     g_string_append_len(info->buf, buffer, bufferSize);
 394     return S3StatusOK;
 395 }
 396
 397 static int s3store_put_handler(int bufferSize, char *buffer,
 398                                void *callbackData)
 399 {
 400     struct put_info *info = (struct put_info *)callbackData;
 401     gint bytes = MIN(bufferSize, (int)(info->val->len - info->offset));
 402     memcpy(buffer, (char *)info->val->data + info->offset, bytes);
 403     info->offset += bytes;
 404     return bytes;
 405 }
 406
 407 static S3Status s3store_properties_callback(const S3ResponseProperties *properties,
 408                                      void *callbackData)
 409 {
 410     return S3StatusOK;
 411 }
 412
 413 static void s3store_response_callback(S3Status status,
 414                                const S3ErrorDetails *errorDetails,
 415                                void *callbackData)
 416 {
 417     struct get_info *info = (struct get_info *)callbackData;
 418
 419     if (status == 0) {
 420         info->success = 1;
 421     }
 422
 423     if (errorDetails != NULL && errorDetails->message != NULL) {
 424         g_print("  Error message: %s\n", errorDetails->message);
 425     }
 426 }
 427
 428 static void s3store_task(gpointer a, gpointer s)
 429 {
 430     BlueSkyStoreAsync *async = (BlueSkyStoreAsync *)a;
 431     S3Store *store = (S3Store *)s;
 432
 433     async->status = ASYNC_RUNNING;
 434     async->exec_time = bluesky_now_hires();
 435
 436     if (async->op == STORE_OP_GET) {
 437         struct get_info info;
 438         info.buf = g_string_new("");
 439         info.success = 0;
 440
 441         struct S3GetObjectHandler handler;
 442         handler.responseHandler.propertiesCallback = s3store_properties_callback;
 443         handler.responseHandler.completeCallback = s3store_response_callback;
 444         handler.getObjectDataCallback = s3store_get_handler;
 445
 446         S3_get_object(&store->bucket, async->key, NULL,
 447                       async->start, async->len, NULL, &handler, &info);
 448         async->range_done = TRUE;
 449
 450         if (info.success) {
 451             async->data = bluesky_string_new_from_gstring(info.buf);
 452             async->result = 0;
 453         } else {
 454             g_string_free(info.buf, TRUE);
 455         }
 456
 457     } else if (async->op == STORE_OP_PUT) {
 458         struct put_info info;
 459         info.success = 0;
 460         info.val = async->data;
 461         info.offset = 0;
 462
 463         struct S3PutObjectHandler handler;
 464         handler.responseHandler.propertiesCallback
 465             = s3store_properties_callback;
 466         handler.responseHandler.completeCallback = s3store_response_callback;
 467         handler.putObjectDataCallback = s3store_put_handler;
 468
 469         S3_put_object(&store->bucket, async->key, async->data->len, NULL, NULL,
 470                       &handler, &info);
 471
 472         if (info.success) {
 473             async->result = 0;
 474         } else {
 475             g_warning("Error completing S3 put operation; client must retry!");
 476         }
 477     }
 478
 479     bluesky_store_async_mark_complete(async);
 480     bluesky_store_async_unref(async);
 481 }
 482
 483 static S3Status s3store_list_handler(int isTruncated,
 484                                      const char *nextMarker,
 485                                      int contentsCount,
 486                                      const S3ListBucketContent *contents,
 487                                      int commonPrefixesCount,
 488                                      const char **commonPrefixes,
 489                                      void *callbackData)
 490 {
 491     struct list_info *info = (struct list_info *)callbackData;
 492     if (contentsCount > 0) {
 493         g_free(info->last_entry);
 494         info->last_entry = g_strdup(contents[contentsCount - 1].key);
 495     }
 496     info->truncated = isTruncated;
 497     return S3StatusOK;
 498 }
 499
 500 static char *s3store_lookup_last(gpointer s, const char *prefix)
 501 {
 502     S3Store *store = (S3Store *)s;
 503     struct list_info info = {0, NULL, FALSE};
 504
 505     struct S3ListBucketHandler handler;
 506     handler.responseHandler.propertiesCallback
 507         = s3store_properties_callback;
 508     handler.responseHandler.completeCallback = s3store_response_callback;
 509     handler.listBucketCallback = s3store_list_handler;
 510
 511     char *marker = NULL;
 512
 513     do {
 514         S3_list_bucket(&store->bucket, prefix, marker, NULL, 1024, NULL,
 515                        &handler, &info);
 516         g_free(marker);
 517         marker = g_strdup(info.last_entry);
 518         g_print("Last key: %s\n", info.last_entry);
 519     } while (info.truncated);
 520
 521     g_free(marker);
 522
 523     return info.last_entry;
 524 }
 525
 526 static gpointer s3store_new(const gchar *path)
 527 {
 528     S3Store *store = g_new(S3Store, 1);
 529     store->thread_pool = g_thread_pool_new(s3store_task, store, -1, FALSE,
 530                                            NULL);
 531     if (path == NULL || strlen(path) == 0)
 532         store->bucket.bucketName = "mvrable-bluesky";
 533     else
 534         store->bucket.bucketName = g_strdup(path);
 535     store->bucket.protocol = S3ProtocolHTTP;
 536     store->bucket.uriStyle = S3UriStylePath;
 537     store->bucket.accessKeyId = getenv("AWS_ACCESS_KEY_ID");
 538     store->bucket.secretAccessKey = getenv("AWS_SECRET_ACCESS_KEY");
 539
 540     const char *key = getenv("BLUESKY_KEY");
 541     if (key == NULL) {
 542         g_error("Encryption key not defined; please set BLUESKY_KEY environment variable");
 543         exit(1);
 544     }
 545
 546     bluesky_crypt_hash_key(key, store->encryption_key);
 547
 548     g_print("Initializing S3 with bucket %s, access key %s, encryption key %s\n",
 549             store->bucket.bucketName, store->bucket.accessKeyId, key);
 550
 551     return store;
 552 }
 553
 554 static void s3store_destroy(gpointer store)
 555 {
 556     g_free(store);
 557 }
 558
 559 static void s3store_submit(gpointer s, BlueSkyStoreAsync *async)
 560 {
 561     S3Store *store = (S3Store *)s;
 562     g_return_if_fail(async->status == ASYNC_NEW);
 563     g_return_if_fail(async->op != STORE_OP_NONE);
 564
 565     switch (async->op) {
 566     case STORE_OP_GET:
 567     case STORE_OP_PUT:
 568         async->status = ASYNC_PENDING;
 569         bluesky_store_async_ref(async);
 570         g_thread_pool_push(store->thread_pool, async, NULL);
 571         break;
 572
 573     default:
 574         g_warning("Uknown operation type for S3Store: %d\n", async->op);
 575         bluesky_store_async_mark_complete(async);
 576         break;
 577     }
 578 }
 579
 580 static void s3store_cleanup(gpointer store, BlueSkyStoreAsync *async)
 581 {
 582     GString *buf = (GString *)async->store_private;
 583
 584     if (buf != NULL) {
 585         g_string_free(buf, TRUE);
 586         async->store_private = NULL;
 587     }
 588 }
 589
 590 static BlueSkyStoreImplementation store_impl = {
 591     .create = s3store_new,
 592     .destroy = s3store_destroy,
 593     .submit = s3store_submit,
 594     .cleanup = s3store_cleanup,
 595     .lookup_last = s3store_lookup_last,
 596 };
 597
 598 void bluesky_store_init_s3(void)
 599 {
 600     S3_initialize(NULL, S3_INIT_ALL);
 601     bluesky_store_register(&store_impl, "s3");
 602 }
 603 #endif