wip on proxy

server.cc

@@ -0,0 +1,668 @@
+// vim: set ts=2 sw=2 et tw=80:
+
+#include <cctype>
+#include <arpa/inet.h>
+#include <Magick++.h>
+#include <sys/types.h>
+#include <cstdint>
+#include <sys/socket.h>
+#include <cstring>
+#include <cstdio>
+#include <cstdlib>
+#include <unistd.h>
+#include <pthread.h>
+#include <iostream>
+#include <map>
+#include <vector>
+#include <curl/curl.h>
+#include <netdb.h>
+
+using namespace std;
+
+#define MAX_CLIENTS 128
+#define BUFFER 8192
+
+void* request_thread(void*);
+
+// Do not kill process when socket connection is broken. Error is already
+// handled
+static void broken_pipe(int signo) {}
+
+int sock_conn_d;
+
+// Close socket when shutting down
+static void sigint(int signo) {
+  close(sock_conn_d);
+  exit(0);
+}
+
+int main(int argc, char** argv) {
+  signal(SIGPIPE, broken_pipe);
+  Magick::InitializeMagick(*argv);
+
+  if(argc < 3) {
+    fprintf(stderr, "Give port and image_rotation\n");
+    exit(255);
+  }
+
+  int sock_data_d;
+  struct sockaddr_in server;
+
+  if((sock_conn_d = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
+    perror("server socket not created");
+    exit(1);
+  }
+
+  memset(&server, 0, sizeof(server)); // zero the memory
+
+  server.sin_family = AF_INET;
+  server.sin_addr.s_addr = inet_addr("0.0.0.0");
+  server.sin_port = htons(atoi(argv[1]));
+
+  int reusePort = 1;
+  setsockopt(sock_conn_d, SOL_SOCKET, SO_REUSEPORT, &reusePort,
+      sizeof(reusePort));
+
+  if(::bind(sock_conn_d, (struct sockaddr*)&server, sizeof(server)) == -1) {
+    perror("server socket not bound");
+    exit(2);
+  }
+
+  while(true) {
+    if(listen(sock_conn_d, MAX_CLIENTS) != 0) {
+      perror("cannot listen on socket");
+      close(sock_conn_d);
+      exit(3);
+    }
+
+    printf("listening...\n");
+
+    struct sockaddr_in client;
+    socklen_t client_len = sizeof(client);
+    sock_data_d = accept(sock_conn_d, (struct sockaddr*)&client, &client_len);
+
+    if(sock_data_d == -1) {
+      perror("cannot accept client");
+      close(sock_conn_d);
+      exit(4);
+    }
+
+    int* fd = (int*) malloc(sizeof(int));
+    if (!fd) {
+      perror("thread creation failed");
+      exit(1);
+    }
+
+    *fd = sock_data_d;
+
+    pthread_t req;
+    pthread_create(&req, NULL, request_thread, (void*) fd);
+  }
+}
+
+bool empty_line(FILE* in) {
+  char a = fgetc(in), b = fgetc(in);
+  if (a == '\r' && b == '\n') {
+    return true;
+  } else {
+    ungetc(b, in);
+    ungetc(a, in);
+    return false;
+  }
+}
+
+bool parse_headers(FILE* in, map<string, string>& headers) {
+  while(true) {
+    // probe first two chars to see if body will start
+    if (empty_line(in)) {
+      break;
+    }
+
+    string header_name;
+    char c;
+
+    while((c = fgetc(in)) != ':') {
+      if (c == EOF || c == '\r' || c == '\n' || !isascii(c)) {
+        return false;
+      } else {
+        header_name += c;
+      }
+    }
+
+    do {
+      c = fgetc(in);
+    } while(isspace(c) && c != '\n' && c != '\r');
+
+    ungetc(c, in);
+
+    string value = "";
+    do {
+      c = fgetc(in);
+      value += c;
+    } while(c != '\r');
+    value.pop_back();
+    fgetc(in); // \n
+
+    headers[header_name] = value;
+    cout << header_name << ": " << value << endl;
+  }
+
+  return true;
+}
+
+bool has_body(char* method) {
+  return !strcmp(method, "POST") || !strcmp(method, "PUT");
+}
+
+bool is_chunk_start(char* buf) {
+  char* i;
+  for (i = buf; isdigit(*i); i++);
+  if (*i != '\r') return false;
+  i++;
+  if (*i != '\n') return false;
+  i++;
+  return *i == '\0';
+}
+
+bool fetch_body(FILE* in, vector<uint8_t>& body,
+    const map<string, string> headers) {
+  bool chunked;
+  size_t length, r = 0;
+
+  auto p = headers.find("Content-Length");
+  chunked = p == headers.end();
+
+  if (!chunked) {
+    length = atol(p->second.c_str());
+  } else {
+    auto q = headers.find("Transfer-Encoding");
+    if (q == headers.end() || q->second != "chunked") {
+      return false;
+    }
+  }
+
+  if (!chunked) {
+    // read Content-Length bytes
+    uint8_t buf[8192];
+    const size_t n = atol(p->second.c_str());
+    for(size_t w = 0; w < n;) {
+      ssize_t r = fread(buf, 1, (n - w) > 8192 ? 8192 : (n - w), in);
+      if (r == -1) return false;
+      body.insert(end(body), begin(buf), end(buf));
+      w += r;
+    }
+  } else {
+    // This was implemented before Prof. Carzaniga said chunked encoding is not
+    // required. I am leaving this just because it is already done.
+    // Read chunks and search for final chunk 0\r\n\r\n
+    char buf[8193];
+    bool chunk_start = true;
+
+    while (true) {
+      fgets(buf, 8193, in);
+
+      if (!strcmp(buf, "0\r\n") && chunk_start) {
+        break;
+      }
+
+      if (chunk_start && is_chunk_start(buf)) {
+        continue;
+      }
+
+      body.insert(end(body), begin(buf), end(buf));
+      body.pop_back();
+
+      char b = body.back();
+      body.pop_back();
+      char a = body.back();
+      body.pop_back();
+      if (a != '\r' || b != '\n') {
+        body.push_back(a);
+        body.push_back(b);
+      }
+
+      size_t len = strlen(buf);
+      chunk_start = len > 2 && buf[len-2] == '\r' && buf[len-1] == '\n';
+    }
+
+    body.insert(end(body), begin(buf), end(buf));
+    body.push_back(fgetc(in)); // \r
+    body.push_back(fgetc(in)); // \n
+  }
+
+  return true;
+}
+
+void send_error(FILE* in, FILE* out, const char* protocol, const int status,
+    const string message) {
+  const char* msg = message.c_str();
+
+  fprintf(out, "%s %d %s\r\n", protocol, status, msg);
+  fprintf(out, "Content-Type: text/html; charset=utf-8\r\n");
+  fprintf(out, "Connection: close\r\n");
+
+  cerr << status << ' ' << msg << endl;
+
+  char message_buf[4096];
+  message_buf[4095] = '\0';
+  snprintf(message_buf, 4095, "<!DOCTYPE html>\n"
+      "<html>\n"
+      "  <head>\n"
+      "    <meta charset=\"utf-8\"/>\n"
+      "    <title>%s</title>\n"
+      "  </head>\n"
+      "  <body>\n"
+      "    <h1>%s</h1>\n"
+      "  </body>\n"
+      "</html>\n", msg, msg);
+
+  fprintf(out, "Content-Length: %lu\r\n\r\n", strlen(message_buf));
+  fprintf(out, "%s", message_buf);
+
+  fclose(in);
+  fflush(out);
+  fclose(out);
+  pthread_exit(NULL);
+}
+
+bool set_curl_method(CURL* c, const char* method) {
+  if (!strcmp(method, "GET")) {
+    curl_easy_setopt(c, CURLOPT_HTTPGET, 1);
+  } else if (!strcmp(method, "HEAD")) {
+    curl_easy_setopt(c, CURLOPT_HTTPGET, 1);
+    curl_easy_setopt(c, CURLOPT_NOBODY, 1);
+  } else if (!strcmp(method, "POST")) {
+    curl_easy_setopt(c, CURLOPT_POST, 1);
+  } else if (!strcmp(method, "PUT")) {
+    curl_easy_setopt(c, CURLOPT_PUT, 1);
+  } else if (!strcmp(method, "DELETE")) {
+    curl_easy_setopt(c, CURLOPT_CUSTOMREQUEST, "DELETE");
+  } else {
+    //TODO: implement CONNECT method
+    return false;
+  }
+
+  return true;
+}
+
+struct buffer {
+  size_t size;
+  uint8_t* data;
+};
+
+void buffer_init(struct buffer* self) {
+  self->size = 0;
+  self->data = (uint8_t*) malloc(8192);
+  if (self->data) {
+    self->data[0] = 0;
+  }
+}
+
+size_t write_buffer(void *ptr, size_t size, size_t nmemb,
+    struct buffer* rbody) {
+  size_t index = rbody->size;
+  size_t n = size * nmemb;
+
+  rbody->size += n;
+
+  uint8_t* newdata = (uint8_t*) realloc(rbody->data, rbody->size + 1);
+
+  if (!newdata) {
+    perror("reallocation of body buffer failed");
+    if (rbody->data) free(rbody->data);
+    rbody->data = NULL;
+    return 0;
+  } else {
+    rbody->data = newdata;
+  }
+
+  memcpy((rbody->data + index), ptr, n);
+  rbody->data[rbody->size] = '\0';
+
+  cout << "copy " << n << " size " << rbody->size << endl;
+
+  return size * nmemb;
+}
+
+// Converts url in a zero-terminated string for the host name. Port is retured
+// as integer, -1 on error
+int find_host_port(char* url) {
+  char* i = url;
+  size_t c = 0;
+
+  while (*i != ':' && *i != '\0') {
+    if (c >= 256 || (!isalnum(*i) && *i != '-' && *i != '.')) return -1;
+    i++;
+    c++;
+  }
+
+  if (c == 0) return -1;
+
+  *i = '\0';
+  i++;
+  if (*i == '\0') return 80;
+
+  unsigned p;
+  if (sscanf(i, "%5u", &p) == 1) {
+    return p < 65536 ? p : 80;
+  } else {
+    return 80;
+  }
+}
+
+struct forward {
+  int in;
+  int out;
+};
+
+
+bool total_write(uint8_t* data, size_t n, int out) {
+  for(size_t w = 0; w < n;) {
+    ssize_t written = write(out, data + w, n - w);
+    if (written == -1) {
+      return false;
+    }
+    w += written;
+  }
+  return true;
+}
+
+void* forwarder_thread(void* data) {
+  struct forward* f = (struct forward*) data;
+  const size_t BUF_SIZE = 65536;
+
+  uint8_t buffer[BUF_SIZE];
+  while (true) {
+    size_t r = read(f->in, buffer, BUF_SIZE);
+
+    if (r == 0) {
+      break;
+    }
+
+    if (!total_write(buffer, r, f->out)) {
+      cout << "Closing CONNECT forwaredr" << endl;
+      return NULL;
+    }
+  }
+
+  return NULL;
+}
+
+void handle_connect(FILE* in, FILE* out, const char* protocol, char* url,
+    const int fd) {
+  int port;
+  if ((port = find_host_port(url)) == -1) {
+    send_error(in, out, protocol, 500, "Hostname parse error");
+  }
+
+  int socketfd = socket(AF_INET, SOCK_STREAM, 0);
+  if (socketfd == -1) {
+    send_error(in, out, protocol, 500, "TCP socket connection error");
+  }
+
+  struct hostent *he = gethostbyname(url);
+  if (!he) {
+    send_error(in, out, protocol, 404, "Unknown host");
+  }
+
+  struct sockaddr_in locale;
+  memset (&locale, 0, sizeof(struct sockaddr_in));
+
+  locale.sin_family =	AF_INET;
+  locale.sin_addr.s_addr	=	htonl(INADDR_ANY);
+  locale.sin_port =	htons(0);
+
+  if (::bind(socketfd, (struct sockaddr*) &locale, sizeof(locale)) == -1) {
+    send_error(in, out, protocol, 500, "TCP socket connection error");
+  }
+
+  struct sockaddr_in remote;
+  memset (&remote, 0, sizeof(struct sockaddr_in));
+
+  remote.sin_family = he->h_addrtype;
+  memmove(&remote.sin_addr, he->h_addr, he->h_length);
+  remote.sin_port = htons((uint16_t) port);
+
+  if (connect(socketfd, (struct sockaddr*) &remote, sizeof(remote)) == -1) {
+    send_error(in, out, protocol, 500, "TCP socket connection error");
+  }
+
+  struct forward from = { .in = fd, .out = socketfd },
+                 to = { .in = socketfd, .out = fd };
+
+
+  fprintf(out, "%s %d %s\r\n\r\n", protocol, 200, "Connection established");
+  fflush(out);
+
+  cout << "Connection established" << endl;
+
+  pthread_t ffrom;
+  pthread_create(&ffrom, NULL, forwarder_thread, &from);
+
+  forwarder_thread(&to);
+
+  pthread_join(ffrom, NULL);
+}
+
+struct read_buffer {
+  const vector<uint8_t>& body;
+  size_t index;
+};
+
+static size_t read_buffer(void *dest, size_t size, size_t nmemb, void *userp)
+{
+  struct read_buffer *wt = (struct read_buffer*) userp;
+  size_t buffer_size = size * nmemb;
+  size_t remaining = wt->body.size() - wt->index;
+
+  if (buffer_size == wt->body.size()) {
+    return 0;
+  }
+
+  size_t chunk_len = remaining > buffer_size ? buffer_size : remaining;
+  auto start = wt->body.begin() + wt->index;
+
+  copy(start, start + chunk_len, (uint8_t*) dest);
+  wt->index += chunk_len;
+  return chunk_len;
+}
+
+void* request_thread(void* data) {
+  int fd = *((int*) data);
+  FILE* in = fdopen(dup(*((int*) data)), "r");
+  FILE* out = fdopen(dup(*((int*) data)), "w");
+  free(data);
+
+  while (true) {
+
+    char method[10];
+    char url[8000];
+    char protocol[10];
+    map<string, string> headers;
+    vector<uint8_t> body;
+
+    if (fscanf(in, "%10s %8000s %10s\r\n", method, url, protocol) < 3) {
+      send_error(in, out, protocol, 400, "Bad request line");
+    }
+
+    cout << "METHOD: " << method << " URL: " << url << " PROTOCOL: " << protocol
+      << endl;
+
+    if (!parse_headers(in, headers)) {
+      send_error(in, out, protocol, 400, "Malformed header");
+    }
+
+    if (strncmp(url, "http", 4) && strcmp(method, "CONNECT")) {
+      send_error(in, out, protocol, 400, "Request line must contain absolute"
+          " URL (this is a proxy)");
+    }
+
+    cout << "BODY: " << endl;
+
+    // check if Content-Length is present and read Content-Length bytes
+    // check if Transfer-Encoding: chunked is present and wait for \r\n on a
+    // line
+
+    if (has_body(method)) {
+      if (!fetch_body(in, body, headers)) {
+        send_error(in, out, protocol, 411,
+            "No Content-Length or Transfer-Encoding");
+      }
+    }
+
+    cout << "PARSED" << endl;
+
+    if (!strcmp(method, "CONNECT")) {
+      handle_connect(in, out, protocol, url, fd);
+      break;
+    } else {
+      struct buffer rhead;
+      struct buffer rbody;
+      buffer_init(&rhead);
+      buffer_init(&rbody);
+
+      if(!rhead.data || !rbody.data) {
+        send_error(in, out, protocol, 500, "Failed to allocate response buffer");
+      }
+
+      CURL *curl;
+      curl = curl_easy_init();
+
+      if (!curl) {
+        send_error(in, out, protocol, 500, "Cannot init CURL");
+      }
+
+      if (!set_curl_method(curl, method)) {
+        send_error(in, out, protocol, 405, "Method not implemented");
+      }
+
+      curl_easy_setopt(curl, CURLOPT_URL, url);
+
+      // Delete Proxy-Connection header
+      {
+        auto i = headers.find("Proxy-Connection");
+        if (i != headers.end()) {
+          headers.erase(i);
+        }
+      }
+
+      struct curl_slist *chunk = NULL;
+      for (auto i = headers.begin(); i != headers.end(); i++) {
+        string header = i->first + ": " + i->second;
+        chunk = curl_slist_append(chunk, header.c_str());
+      }
+      curl_easy_setopt(curl, CURLOPT_HTTPHEADER, chunk);
+
+      if (has_body(method)) {
+        struct read_buffer r = { .body = body, .index = 0 };
+        curl_easy_setopt(curl, CURLOPT_READFUNCTION, read_buffer);
+        curl_easy_setopt(curl, CURLOPT_READDATA, &r);
+
+        chunk = curl_slist_append(chunk, "Transfer-Encoding: chunked");
+        chunk = curl_slist_append(chunk, "Expect:");
+      }
+
+      curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, write_buffer);
+      curl_easy_setopt(curl, CURLOPT_HEADERDATA, &rhead);
+      curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_buffer);
+      curl_easy_setopt(curl, CURLOPT_WRITEDATA, &rbody);
+      CURLcode res = curl_easy_perform(curl);
+      if(res != CURLE_OK) {
+        fprintf(stderr, "curl_easy_perform() failed: %s\n",
+            curl_easy_strerror(res));
+        send_error(in, out, protocol, 502, "Request to server failed");
+      } else {
+        char protocol_r[10];
+        char message[8000];
+        unsigned code;
+
+        const char* image = NULL;
+        const char* const PNG = "PNG";
+        const char* const JPEG = "JPEG";
+
+        {
+          cout << "RESPONSE: " << endl;
+          FILE* res_head = fmemopen(rhead.data, rhead.size, "r");
+
+          fscanf(res_head, "%10s %3u %8000s\r\n", protocol_r, &code, message);
+
+          cout << "STATUS: " << code << " PROTOCOL: " << protocol_r << endl;
+
+          headers.empty();
+          parse_headers(res_head, headers);
+
+          auto a = headers.find("Content-Type");
+          if (a != headers.end()) {
+            if (a->second == "image/jpeg") image = JPEG;
+            else if (a->second == "image/png") image = PNG;
+          }
+        }
+
+        cout << "image: " << (image ? image : "NULL") << endl;
+
+
+        if (image) {
+          try {
+            Magick::Blob my_blob(rbody.data, rbody.size);
+            Magick::Blob output;
+
+            Magick::Image to_rotate(my_blob);
+            to_rotate.magick(image);
+            to_rotate.flip();
+            to_rotate.write(&output);
+
+            cout << "IN" << endl << my_blob.length() << endl;
+            cout << "OUT" << endl << output.length() << endl;
+
+            fprintf(out, "%s %u %s\n", protocol_r, code, message);
+            cout << protocol_r << ' ' << code << ' ' << message << endl;
+
+            headers["Content-Length"] = to_string(output.length());
+            for (auto i = headers.begin(); i != headers.end(); i++) {
+              string header = i->first + ": " + i->second + "\r\n";
+              fprintf(out, "%s\n", header.c_str());
+              cout << header << endl;
+            }
+            fprintf(out, "\r\n");
+            fflush(out);
+
+            total_write((uint8_t*) output.data(), output.length(), fd);
+          } catch (Magick::Exception &error) {
+            send_error(in, out, protocol, 500, "Image conversion failed");
+            fprintf(stderr, "I exit %s\n", error.what());
+            break;
+          } catch (std::exception &error) {
+            send_error(in, out, protocol, 500, "Image conversion failed");
+            fprintf(stderr, "I exit %s\n", error.what());
+            break;
+          }
+        } else {
+          total_write(rhead.data, rhead.size, fd);
+          total_write(rbody.data, rbody.size, fd);
+          fflush(out);
+        }
+      }
+
+      curl_easy_cleanup(curl);
+      curl_slist_free_all(chunk);
+      free(rhead.data);
+      free(rbody.data);
+    }
+
+    {
+      auto i = headers.find("Connection");
+      if (i != headers.end() && i->second == "close") {
+        break;
+      }
+    }
+  }
+
+  cout << "closing data socket...\n";
+
+  fclose(in);
+
+  fflush(out);
+  fclose(out);
+
+  pthread_exit(NULL);
+}