ostree/src/libostree/ostree-repo-static-delta-processing.c

/* -*- mode: C; c-file-style: "gnu"; indent-tabs-mode: nil; -*-
 *
 * Copyright (C) 2013,2014 Colin Walters <walters@verbum.org>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 02111-1307, USA.
 */

#include "config.h"

#include <string.h>

#include <glib-unix.h>
#include <gio/gunixinputstream.h>
#include <gio/gunixoutputstream.h>
#include <gio/gfiledescriptorbased.h>

#include "ostree-core-private.h"
#include "ostree-repo-private.h"
#include "ostree-repo-static-delta-private.h"
#include "ostree-lzma-decompressor.h"
#include "otutil.h"
#include "ostree-varint.h"

/* This should really always be true, but hey, let's just assert it */
G_STATIC_ASSERT (sizeof (guint) >= sizeof (guint32));

typedef struct {
  OstreeRepo     *repo;
  guint           checksum_index;
  const guint8   *checksums;
  guint           n_checksums;

  const guint8   *opdata;
  guint           oplen;

  GVariant       *mode_dict;
  GVariant       *xattr_dict;
  
  gboolean        object_start;
  gboolean        caught_error;
  GError        **async_error;

  OstreeObjectType output_objtype;
  const guint8   *output_target;
  const guint8   *input_target_csum;

  const guint8   *payload_data;
  guint64         payload_size; 
} StaticDeltaExecutionState;

typedef struct {
  StaticDeltaExecutionState *state;
  char checksum[65];
} StaticDeltaContentWrite;

typedef gboolean (*DispatchOpFunc) (OstreeRepo                 *repo,
                                    StaticDeltaExecutionState  *state,
                                    GCancellable               *cancellable,
                                    GError                    **error);

typedef struct  {
  const char *name;
  DispatchOpFunc func;
} OstreeStaticDeltaOperation;

#define OPPROTO(name) \
  static gboolean dispatch_##name (OstreeRepo                 *repo, \
                                   StaticDeltaExecutionState  *state, \
                                   GCancellable               *cancellable, \
                                   GError                    **error);

OPPROTO(open_splice_and_close)
OPPROTO(open)
OPPROTO(write)
OPPROTO(set_read_source)
OPPROTO(close)
#undef OPPROTO

static gboolean
read_varuint64 (StaticDeltaExecutionState  *state,
                guint64                    *out_value,
                GError                    **error)
{
  gsize bytes_read;
  if (!_ostree_read_varuint64 (state->opdata, state->oplen, out_value, &bytes_read))
    {
      g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_FAILED,
                           "Unexpected EOF reading varint");
      return FALSE;
    }
  state->opdata += bytes_read;
  state->oplen -= bytes_read;
  return TRUE;
}

static gboolean
open_output_target (StaticDeltaExecutionState   *state,
                    GCancellable                *cancellable,
                    GError                     **error)
{
  gboolean ret = FALSE;
  guint8 *objcsum;
  char checksum[65];

  g_assert (state->checksums != NULL);
  g_assert (state->output_target == NULL);
  g_assert (state->checksum_index < state->n_checksums);

  objcsum = (guint8*)state->checksums + (state->checksum_index * OSTREE_STATIC_DELTA_OBJTYPE_CSUM_LEN);

  if (G_UNLIKELY(!ostree_validate_structureof_objtype (*objcsum, error)))
    goto out;

  state->output_objtype = (OstreeObjectType) *objcsum;
  state->output_target = objcsum + 1;

  ostree_checksum_inplace_from_bytes (state->output_target, checksum);

  ret = TRUE;
 out:
  return ret;
}

gboolean
_ostree_static_delta_part_validate (OstreeRepo     *repo,
                                    GFile          *part_path,
                                    guint           part_offset,
                                    const char     *expected_checksum,
                                    GCancellable   *cancellable,
                                    GError        **error)
{
  gboolean ret = FALSE;
  gs_unref_object GInputStream *tmp_in = NULL;
  gs_free guchar *actual_checksum_bytes = NULL;
  gs_free gchar *actual_checksum = NULL;
  
  tmp_in = (GInputStream*)g_file_read (part_path, cancellable, error);
  if (!tmp_in)
    goto out;
  
  if (!ot_gio_checksum_stream (tmp_in, &actual_checksum_bytes,
                               cancellable, error))
    goto out;

  actual_checksum = ostree_checksum_from_bytes (actual_checksum_bytes);
  if (strcmp (actual_checksum, expected_checksum) != 0)
    {
      g_set_error (error, G_IO_ERROR, G_IO_ERROR_FAILED,
                   "Checksum mismatch in static delta part %u; expected=%s actual=%s",
                   part_offset, expected_checksum, actual_checksum);
      goto out;
    }
  
  ret = TRUE;
 out:
  return ret;
}

gboolean
_ostree_static_delta_part_execute_raw (OstreeRepo      *repo,
                                       GVariant        *objects,
                                       GVariant        *part,
                                       GCancellable    *cancellable,
                                       GError         **error)
{
  gboolean ret = FALSE;
  guint8 *checksums_data;
  gs_unref_variant GVariant *checksums = NULL;
  gs_unref_variant GVariant *mode_dict = NULL;
  gs_unref_variant GVariant *xattr_dict = NULL;
  gs_unref_variant GVariant *payload = NULL;
  gs_unref_variant GVariant *ops = NULL;
  StaticDeltaExecutionState statedata = { 0, };
  StaticDeltaExecutionState *state = &statedata;
  guint n_executed = 0;

  state->repo = repo;
  state->async_error = error;

  if (!_ostree_static_delta_parse_checksum_array (objects,
                                                  &checksums_data,
                                                  &state->n_checksums,
                                                  error))
    goto out;

  state->checksums = checksums_data;
  g_assert (state->n_checksums > 0);

  g_variant_get (part, "(@a(uuu)@aa(ayay)@ay@ay)",
                 &mode_dict,
                 &xattr_dict,
                 &payload, &ops);

  state->mode_dict = mode_dict;
  state->xattr_dict = xattr_dict;

  state->payload_data = g_variant_get_data (payload);
  state->payload_size = g_variant_get_size (payload);

  state->oplen = g_variant_n_children (ops);
  state->opdata = g_variant_get_data (ops);

  while (state->oplen > 0)
    {
      guint8 opcode;

      opcode = state->opdata[0];
      state->oplen--;
      state->opdata++;

      switch (opcode)
        {
        case OSTREE_STATIC_DELTA_OP_OPEN_SPLICE_AND_CLOSE:
          if (!dispatch_open_splice_and_close (repo, state, cancellable, error))
            goto out;
          break;
        case OSTREE_STATIC_DELTA_OP_OPEN:
          if (!dispatch_open (repo, state, cancellable, error))
            goto out;
          break;
        case OSTREE_STATIC_DELTA_OP_WRITE:
          if (!dispatch_write (repo, state, cancellable, error))
            goto out;
          break;
        case OSTREE_STATIC_DELTA_OP_SET_READ_SOURCE:
          if (!dispatch_set_read_source (repo, state, cancellable, error))
            goto out;
          break;
        case OSTREE_STATIC_DELTA_OP_CLOSE:
          if (!dispatch_close (repo, state, cancellable, error))
            goto out;
          break;
        default:
          g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
                       "Unknown opcode %u at offset %u", opcode, n_executed);
          goto out;
        }

      n_executed++;
    }

  if (state->caught_error)
    goto out;

  ret = TRUE;
 out:
  return ret;
}

static gboolean
decompress_all (GConverter   *converter,
                GBytes       *data,
                GBytes      **out_uncompressed,
                GCancellable *cancellable,
                GError      **error)
{
  gboolean ret = FALSE;
  gs_unref_object GMemoryInputStream *memin = (GMemoryInputStream*)g_memory_input_stream_new_from_bytes (data);
  gs_unref_object GMemoryOutputStream *memout = (GMemoryOutputStream*)g_memory_output_stream_new (NULL, 0, g_realloc, g_free);
  gs_unref_object GInputStream *convin = g_converter_input_stream_new ((GInputStream*)memin, converter);

  if (0 > g_output_stream_splice ((GOutputStream*)memout, convin,
                                  G_OUTPUT_STREAM_SPLICE_CLOSE_SOURCE |
                                  G_OUTPUT_STREAM_SPLICE_CLOSE_TARGET,
                                  cancellable, error))
    goto out;

  ret = TRUE;
  *out_uncompressed = g_memory_output_stream_steal_as_bytes (memout);
 out:
  return ret;
}

gboolean
_ostree_static_delta_part_execute (OstreeRepo      *repo,
                                   GVariant        *header,
                                   GBytes          *part_bytes,
                                   GCancellable    *cancellable,
                                   GError         **error)
{
  gboolean ret = FALSE;
  gsize partlen;
  const guint8*partdata;
  gs_unref_bytes GBytes *part_payload_bytes = NULL;
  gs_unref_bytes GBytes *payload_data = NULL;
  gs_unref_variant GVariant *payload = NULL;
  guint8 comptype;

  partdata = g_bytes_get_data (part_bytes, &partlen);
  
  if (partlen < 1)
    {
      g_set_error (error, G_IO_ERROR, G_IO_ERROR_FAILED,
                   "Corrupted 0 length delta part");
      goto out;
    }
        
  /* First byte is compression type */
  comptype = partdata[0];
  /* Then the rest may be compressed or uncompressed */
  part_payload_bytes = g_bytes_new_from_bytes (part_bytes, 1, partlen - 1);
  switch (comptype)
    {
    case 0:
      /* No compression */
      payload_data = g_bytes_ref (part_payload_bytes);
      break;
    case 'x':
      {
        gs_unref_object GConverter *decomp =
          (GConverter*) _ostree_lzma_decompressor_new ();

        if (!decompress_all (decomp, part_payload_bytes, &payload_data,
                             cancellable, error))
          goto out;
      }
      break;
    default:
      g_set_error (error, G_IO_ERROR, G_IO_ERROR_FAILED,
                   "Invalid compression type '%u'", comptype);
      goto out;
    }
        
  payload = ot_variant_new_from_bytes (G_VARIANT_TYPE (OSTREE_STATIC_DELTA_PART_PAYLOAD_FORMAT_V0),
                                       payload_data, FALSE);
  if (!_ostree_static_delta_part_execute_raw (repo, header, payload,
                                              cancellable, error))
    goto out;

  ret = TRUE;
 out:
  return ret;
}

typedef struct {
  OstreeRepo *repo;
  GVariant *header;
  GBytes *partdata;
  GCancellable *cancellable;
  GSimpleAsyncResult *result;
} StaticDeltaPartExecuteAsyncData;

static void
static_delta_part_execute_async_data_free (gpointer user_data)
{
  StaticDeltaPartExecuteAsyncData *data = user_data;

  g_clear_object (&data->repo);
  g_variant_unref (data->header);
  g_bytes_unref (data->partdata);
  g_clear_object (&data->cancellable);
  g_free (data);
}

static void
static_delta_part_execute_thread (GSimpleAsyncResult  *res,
                                  GObject             *object,
                                  GCancellable        *cancellable)
{
  GError *error = NULL;
  StaticDeltaPartExecuteAsyncData *data;

  data = g_simple_async_result_get_op_res_gpointer (res);
  if (!_ostree_static_delta_part_execute (data->repo,
                                          data->header,
                                          data->partdata,
                                          cancellable, &error))
    g_simple_async_result_take_error (res, error);
}

void
_ostree_static_delta_part_execute_async (OstreeRepo      *repo,
                                         GVariant        *header,
                                         GBytes          *partdata,
                                         GCancellable    *cancellable,
                                         GAsyncReadyCallback  callback,
                                         gpointer         user_data)
{
  StaticDeltaPartExecuteAsyncData *asyncdata;

  asyncdata = g_new0 (StaticDeltaPartExecuteAsyncData, 1);
  asyncdata->repo = g_object_ref (repo);
  asyncdata->header = g_variant_ref (header);
  asyncdata->partdata = g_bytes_ref (partdata);
  asyncdata->cancellable = cancellable ? g_object_ref (cancellable) : NULL;

  asyncdata->result = g_simple_async_result_new ((GObject*) repo,
                                                 callback, user_data,
                                                 _ostree_static_delta_part_execute_async);

  g_simple_async_result_set_op_res_gpointer (asyncdata->result, asyncdata,
                                             static_delta_part_execute_async_data_free);
  g_simple_async_result_run_in_thread (asyncdata->result, static_delta_part_execute_thread, G_PRIORITY_DEFAULT, cancellable);
  g_object_unref (asyncdata->result);
}

gboolean
_ostree_static_delta_part_execute_finish (OstreeRepo      *repo,
                                          GAsyncResult    *result,
                                          GError         **error) 
{
  GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);

  g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == _ostree_static_delta_part_execute_async);

  if (g_simple_async_result_propagate_error (simple, error))
    return FALSE;
  return TRUE;
}

static gboolean
validate_ofs (StaticDeltaExecutionState  *state,
              guint64                     offset,
              guint64                     length,
              GError                    **error)
{
  if (G_UNLIKELY (offset + length < offset ||
                  offset + length > state->payload_size))
    {
      g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
                   "Invalid offset/length %" G_GUINT64_FORMAT "/%" G_GUINT64_FORMAT,
                   offset, length);
      return FALSE;
    }
  return TRUE;
}

static gboolean
dispatch_open_splice_and_close (OstreeRepo                 *repo,
                                StaticDeltaExecutionState  *state,
                                GCancellable               *cancellable,  
                                GError                    **error)
{
  gboolean ret = FALSE;
  char checksum[65];

  if (!open_output_target (state, cancellable, error))
    goto out;

  ostree_checksum_inplace_from_bytes (state->output_target, checksum);

  if (OSTREE_OBJECT_TYPE_IS_META (state->output_objtype))
    {
      gs_unref_variant GVariant *metadata = NULL;
      guint64 offset;
      guint64 length;

      if (!read_varuint64 (state, &length, error))
        goto out;
      if (!read_varuint64 (state, &offset, error))
        goto out;
      if (!validate_ofs (state, offset, length, error))
        goto out;
      
      metadata = g_variant_new_from_data (ostree_metadata_variant_type (state->output_objtype),
                                          state->payload_data + offset, length, TRUE, NULL, NULL);
      
      if (!ostree_repo_write_metadata_trusted (state->repo, state->output_objtype,
                                               checksum,
                                               metadata,
                                               cancellable,
                                               error))
        goto out;
    }
  else
    {
      guint64 mode_offset;
      guint64 xattr_offset;
      guint64 content_size;
      guint64 content_offset;
      guint64 objlen;
      gs_unref_object GInputStream *object_input = NULL;
      gs_unref_object GInputStream *memin = NULL;
      gs_unref_variant GVariant *xattrs_buf = NULL;
      GVariant *modev;
      GVariant *xattrs;
      guint32 uid, gid, mode;

      if (!read_varuint64 (state, &mode_offset, error))
        goto out;
      if (!read_varuint64 (state, &xattr_offset, error))
        goto out;
      if (!read_varuint64 (state, &content_size, error))
        goto out;
      if (!read_varuint64 (state, &content_offset, error))
        goto out;

      if (!validate_ofs (state, content_offset, content_size, error))
        goto out;

      modev = g_variant_get_child_value (state->mode_dict, mode_offset);
      g_variant_get (modev, "(uuu)", &uid, &gid, &mode);
      uid = GUINT32_FROM_BE (uid);
      gid = GUINT32_FROM_BE (gid);
      mode = GUINT32_FROM_BE (mode);

      xattrs = g_variant_get_child_value (state->xattr_dict, xattr_offset);

      /* Fast path for regular files to bare repositories */
      if (S_ISREG (mode) && 
          (repo->mode == OSTREE_REPO_MODE_BARE ||
           repo->mode == OSTREE_REPO_MODE_BARE_USER))
        {
          OstreeRepoTrustedContentBareCommit barecommitstate = { -1 };
          gs_unref_object GOutputStream *outstream = NULL;
          gsize bytes_written;
          gboolean have_obj;

          if (!_ostree_repo_open_trusted_content_bare (repo, checksum,
                                                       content_size,
                                                       &barecommitstate,
                                                       &outstream,
                                                       &have_obj,
                                                       cancellable, error))
            goto out;
          
          if (!have_obj)
            {
              if (!g_output_stream_write_all (outstream, state->payload_data + content_offset,
                                              content_size,
                                              &bytes_written,
                                              cancellable, error))
                goto out;

              if (!g_output_stream_flush (outstream, cancellable, error))
                goto out;
            }

          if (!_ostree_repo_commit_trusted_content_bare (repo, checksum, &barecommitstate,
                                                         uid, gid, mode, xattrs,
                                                         cancellable, error))
            goto out;
        }
      else
        {
          /* Slower path, for symlinks and unpacking deltas into archive-z2 */
          gs_unref_object GFileInfo *finfo = NULL;
      
          finfo = _ostree_header_gfile_info_new (mode, uid, gid);

          if (S_ISLNK (mode))
            {
              gs_free char *nulterminated_target =
                g_strndup ((char*)state->payload_data + content_offset, content_size);
              g_file_info_set_symlink_target (finfo, nulterminated_target);
            }
          else
            {
              g_assert (S_ISREG (mode));
              g_file_info_set_size (finfo, content_size);
              memin = g_memory_input_stream_new_from_data (state->payload_data + content_offset, content_size, NULL);
            }

          if (!ostree_raw_file_to_content_stream (memin, finfo, xattrs,
                                                  &object_input, &objlen,
                                                  cancellable, error))
            goto out;
          
          if (!ostree_repo_write_content_trusted (state->repo,
                                                  checksum,
                                                  object_input,
                                                  objlen,
                                                  cancellable,
                                                  error))
            goto out;
        }
    }

  state->checksum_index++;
  state->output_target = NULL;

  ret = TRUE;
 out:
  if (!ret)
    g_prefix_error (error, "opcode open-splice-and-close: ");
  return ret;
}

static gboolean
dispatch_open (OstreeRepo                 *repo,
               StaticDeltaExecutionState  *state,
               GCancellable               *cancellable,  
               GError                    **error)
{
  gboolean ret = FALSE;

  if (!open_output_target (state, cancellable, error))
    goto out;

  ret = TRUE;
 out:
  if (!ret)
    g_prefix_error (error, "opcode open: ");
  return ret;
}

static gboolean
dispatch_write (OstreeRepo                 *repo,
               StaticDeltaExecutionState  *state,
               GCancellable               *cancellable,  
               GError                    **error)
{
  gboolean ret = FALSE;

  g_assert_not_reached ();
  
  ret = TRUE;
  /* out: */
  if (!ret)
    g_prefix_error (error, "opcode open-splice-and-close: ");
  return ret;
}

static gboolean
dispatch_set_read_source (OstreeRepo                 *repo,
                          StaticDeltaExecutionState  *state,
                          GCancellable               *cancellable,  
                          GError                    **error)
{
  gboolean ret = FALSE;

  g_assert_not_reached ();
  
  ret = TRUE;
  /* out: */
  if (!ret)
    g_prefix_error (error, "opcode open-splice-and-close: ");
  return ret;
}

static gboolean
dispatch_close (OstreeRepo                 *repo,
                StaticDeltaExecutionState  *state,
                GCancellable               *cancellable,  
                GError                    **error)
{
  gboolean ret = FALSE;

  g_assert_not_reached ();
  
  ret = TRUE;
  /* out: */
  if (!ret)
    g_prefix_error (error, "opcode open-splice-and-close: ");
  return ret;
}