root/branches/cu-security-branch/src/server/set-attr.sm @ 8362

/* 
 * (C) 2001 Clemson University and The University of Chicago 
 *
 * See COPYING in top-level directory.
 */

#include <string.h>
#include <assert.h>

#include "server-config.h"
#include "pvfs2-server.h"
#include "pvfs2-attr.h"
#include "pvfs2-util.h"
#include "pvfs2-internal.h"
#include "pint-util.h"
#include "pint-security.h"

enum
{
    STATE_METAFILE = 7,
    STATE_SYMLINK = 8
};

%%

machine pvfs2_set_attr_sm
{
    state prelude
    {
        jump pvfs2_prelude_sm;
        success => verify_attribs;
        default => final_response;
    }

    state verify_attribs
    {
        run setattr_verify_attribs;
        STATE_METAFILE => write_metafile_datafile_handles_if_required;
        STATE_SYMLINK => write_symlink_target_if_required;
        success => setobj_attrib;
        default => final_response;
    }

    state write_metafile_datafile_handles_if_required
    {
        run setattr_write_metafile_datafile_handles_if_required;
        success => write_metafile_distribution_if_required;
        default => final_response;
    }

    state write_metafile_distribution_if_required
    {
        run setattr_write_metafile_distribution_if_required;
        success => setobj_attrib;
        default => final_response;
    }

    state write_symlink_target_if_required
    {
        run setattr_write_symlink_target_if_required;
        success => setobj_attrib;
        default => final_response;
    }

    state setobj_attrib
    {
        run setattr_setobj_attribs;
        default => final_response;
    }

    state final_response
    {
        jump pvfs2_final_response_sm;
        default => cleanup;
    }

    state cleanup
    {
        run setattr_cleanup;
        default => terminate;
    }
}

%%

static PINT_sm_action setattr_verify_attribs(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PVFS_object_attr *a_p = NULL, *req_a_p = NULL;

    a_p = &s_op->attr;
    req_a_p = &s_op->req->u.setattr.attr;

    gossip_debug(GOSSIP_SETATTR_DEBUG, "  attrs read from dspace:\n\t"
                 "[owner = %d, group = %d, perms = %o, type = %d]\n",
                 a_p->owner, a_p->group, a_p->perms, a_p->objtype);

    gossip_debug(GOSSIP_SETATTR_DEBUG, "  attrs read from request:\n\t"
                 "[owner = %d, group = %d, perms = %o, type = %d]\n",
                 (req_a_p->mask & PVFS_ATTR_COMMON_UID)  ? req_a_p->owner : -1,
                 (req_a_p->mask & PVFS_ATTR_COMMON_GID)  ? req_a_p->group : -1,
                 (req_a_p->mask & PVFS_ATTR_COMMON_PERM) ? req_a_p->perms : -1,
                 (req_a_p->mask & PVFS_ATTR_COMMON_TYPE) ? req_a_p->objtype : -1);
    /*
      here we're enforcing that no one can change the type of the
      handle/object already stored once it's been set to a non-zero
      value.  (zero is not a valid object type meaning that it hasn't
      been assigned yet)
    */
    if (a_p->objtype && req_a_p->objtype &&
        (a_p->objtype != req_a_p->objtype))
    {
        gossip_debug(GOSSIP_SETATTR_DEBUG, "  handle %llu is of type %d "
                     "and cannot be changed to type %d\n",
                     llu(s_op->req->u.setattr.handle),
                     a_p->objtype, s_op->req->u.setattr.attr.objtype);

        /* set an error to bail out of set-attr processing */
        js_p->error_code = -PVFS_EACCES;
        return SM_ACTION_COMPLETE;
    }
    else if (req_a_p->objtype == PVFS_TYPE_NONE)
    {
        /* if the requested object type is PVFS_TYPE_NONE, then the
         * setattr is only on the common attributes, so we use the
         * actual object type
         */
        req_a_p->objtype = a_p->objtype;
    }

    js_p->error_code = 0;

    if ((req_a_p->objtype == PVFS_TYPE_METAFILE) ||
        (a_p->objtype == PVFS_TYPE_METAFILE))
    {
        gossip_debug(GOSSIP_SETATTR_DEBUG,
                     "  handle %llu refers to a metafile\n",
                     llu(s_op->req->u.setattr.handle));

        gossip_debug(
            GOSSIP_SETATTR_DEBUG, " *** dspace has dfile count %d and "
            "req has dfile count %d\n",
            a_p->u.meta.dfile_count, req_a_p->u.meta.dfile_count);
        gossip_debug(
            GOSSIP_SETATTR_DEBUG, " *** dspace has dist size %d and "
            "req has dist size %d\n",
            a_p->u.meta.dist_size, req_a_p->u.meta.dist_size);

        /* copy the dfile count before writing this object */
        if (req_a_p->mask & PVFS_ATTR_META_DFILES)
        {
            gossip_debug(
                GOSSIP_SETATTR_DEBUG," *** using dfile_count of %d\n",
                req_a_p->u.meta.dfile_count);
            a_p->u.meta.dfile_count = req_a_p->u.meta.dfile_count;
            js_p->error_code = STATE_METAFILE;
        }
        else
        {
            gossip_debug(GOSSIP_SETATTR_DEBUG,
                         " *** ignoring dfile_count of %d\n",
                         req_a_p->u.meta.dfile_count);
        }

        /* copy the dist size before writing this object */
        if (req_a_p->mask & PVFS_ATTR_META_DIST)
        {
            gossip_debug(GOSSIP_SETATTR_DEBUG,
                         " *** using dist_size of %d\n",
                         req_a_p->u.meta.dist_size);
            a_p->u.meta.dist_size = req_a_p->u.meta.dist_size;
            js_p->error_code = STATE_METAFILE;
        }
        else
        {
            gossip_debug(GOSSIP_SETATTR_DEBUG,
                         " *** ignoring dist_size of %d\n",
                         req_a_p->u.meta.dist_size);
        }
    }
    else if ((req_a_p->objtype == PVFS_TYPE_DATAFILE) ||
             (a_p->objtype == PVFS_TYPE_DATAFILE))
    {
        gossip_debug(GOSSIP_SETATTR_DEBUG,
                     "  handle %llu refers to a datafile\n",
                     llu(s_op->req->u.setattr.handle));
    }
    else if ((req_a_p->objtype == PVFS_TYPE_DIRECTORY) ||
             (a_p->objtype == PVFS_TYPE_DIRECTORY))
    {
        gossip_debug(GOSSIP_SETATTR_DEBUG,
                     "  handle %llu refers to a directory\n",
                     llu(s_op->req->u.setattr.handle));
    }
    else if ((req_a_p->objtype == PVFS_TYPE_SYMLINK) ||
             (a_p->objtype == PVFS_TYPE_SYMLINK))
    {
        gossip_debug(GOSSIP_SETATTR_DEBUG,
                     "  handle %llu refers to a symlink\n",
                     llu(s_op->req->u.setattr.handle));

        if (req_a_p->mask & PVFS_ATTR_SYMLNK_ALL)
        {
            assert(req_a_p->u.sym.target_path_len > 0);
            assert(req_a_p->u.sym.target_path);

            gossip_debug(GOSSIP_SETATTR_DEBUG,
                         " symlink links handle %llu to %s\n",
                         llu(s_op->req->u.setattr.handle),
                         req_a_p->u.sym.target_path);

            a_p->u.sym.target_path_len = req_a_p->u.sym.target_path_len;
            a_p->u.sym.target_path = req_a_p->u.sym.target_path;
        }
        js_p->error_code = STATE_SYMLINK;
    }
    else if ((req_a_p->objtype == PVFS_TYPE_DIRDATA) ||
             (a_p->objtype == PVFS_TYPE_DIRDATA))
    {
        gossip_debug(GOSSIP_SETATTR_DEBUG,
                     "  handle %llu refers to a dirdata object\n",
                     llu(s_op->req->u.setattr.handle));
    }
    else
    {
        gossip_debug(GOSSIP_SETATTR_DEBUG,
                     "  handle %llu refers to something unknown\n",
                     llu(s_op->req->u.setattr.handle));

        js_p->error_code = -PVFS_EACCES;
    }

    /* verify that the user has permission to change the file's owner */
    if (req_a_p->mask & PVFS_ATTR_COMMON_UID)
    {
        if (!(s_op->req->capability.op_mask & PINT_CAP_ADMIN) &&
            (s_op->req->u.setattr.credential.userid != req_a_p->owner))
        {
            js_p->error_code = -PVFS_EPERM;
        }
    }

    /* verify that the user has permission to change the file's group */
    if (req_a_p->mask & PVFS_ATTR_COMMON_GID)
    {
        PVFS_credential *cred = &s_op->req->u.setattr.credential;
        int i;

        if (!(s_op->req->capability.op_mask & PINT_CAP_ADMIN))
        {
            for (i = 0; i < cred->num_groups; i++)
            {
                if (cred->group_array[i] == req_a_p->group)
                {
                    break;
                }
            }
            /* no group matches */
            if (i >= cred->num_groups)
            {
                js_p->error_code = -PVFS_EPERM;
            }
        }
    }

    return SM_ACTION_COMPLETE;
}

static PINT_sm_action setattr_setobj_attribs(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = -1;
    job_id_t j_id;
    PVFS_object_attr *a_p = NULL;
    PVFS_object_attr *dspace_a_p = NULL;
    PVFS_ds_attributes *ds_attr = NULL;

    dspace_a_p = &s_op->attr;
    a_p = &s_op->req->u.setattr.attr;

    if (a_p->mask & PVFS_ATTR_META_DFILES)
    {
        gossip_debug(GOSSIP_SETATTR_DEBUG, " request has dfile_count of "
                     "%d | dspace has %d\n",
                     s_op->req->u.setattr.attr.u.meta.dfile_count,
                     s_op->attr.u.meta.dfile_count);

        gossip_debug(GOSSIP_SETATTR_DEBUG, " writing count of %d to "
                     "disk\n", dspace_a_p->u.meta.dfile_count);
    }
    /* 
     * Remember that mtime is versioned on disk! so convert it here..
     * It is better to do it here than change the PVFS_object_attr_overwrite_setable
     * macro, since there are many more users of it, I think.
     */
     if (a_p->mask & PVFS_ATTR_COMMON_MTIME_SET)
     {
         PVFS_time orig_mtime = a_p->mtime;
         a_p->mtime = PINT_util_mktime_version(orig_mtime);
         gossip_debug(GOSSIP_SETATTR_DEBUG, "setting version "
                 "to %llu\n\tmtime is %llu\n",
                 llu(a_p->mtime), llu(orig_mtime));
     }

    /* if the object is a symbolic link, check to make sure that the request
     * is not attempting to change the permissions 
     */
    if(dspace_a_p->objtype == PVFS_TYPE_SYMLINK)
    {
        if (dspace_a_p->perms != 0 && ((a_p->mask & PVFS_ATTR_COMMON_PERM) && (dspace_a_p->perms != a_p->perms)))
        {
            gossip_debug(GOSSIP_SETATTR_DEBUG, "Cannot change perms of symlink: Permission denied\n");
            js_p->error_code = -PVFS_EPERM;
            return SM_ACTION_COMPLETE;
        }
    }

    /*
      we have the attribs stored in the dspace, as well as the
      requested attribs to store.  overwrite the ones that are setable
      and specified by the mask value in the request; macro defined in
      pvfs2-storage.h
    */
    PVFS_object_attr_overwrite_setable(dspace_a_p, a_p);

    gossip_debug(
        GOSSIP_SETATTR_DEBUG,
        "  WRITING attrs: [owner = %d, group = %d\n\t"
        "perms = %o, type = %d, atime = %llu, mtime = %llu\n\t"
        "ctime = %llu | dfile_count = %d | dist_size = %d]\n",
        dspace_a_p->owner, dspace_a_p->group, dspace_a_p->perms,
        dspace_a_p->objtype, llu(dspace_a_p->atime),
        llu(PINT_util_mkversion_time(dspace_a_p->mtime)), llu(dspace_a_p->ctime),
        (int)dspace_a_p->u.meta.dfile_count,
        (int)dspace_a_p->u.meta.dist_size);

    /* translate attrs to storage attr format */
    ds_attr = &(s_op->ds_attr);
    PVFS_object_attr_to_ds_attr(dspace_a_p, ds_attr);

    ret = job_trove_dspace_setattr(
        s_op->req->u.setattr.fs_id, s_op->req->u.setattr.handle,
        ds_attr, 
        TROVE_SYNC,
        smcb, 0, js_p, &j_id, server_job_context, s_op->req->hints);

    return ret;
}

static PINT_sm_action setattr_write_metafile_datafile_handles_if_required(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = 0, dfile_count = 0;
    job_id_t j_id;

    /* reset from jump to here with STATE_METAFILE */
    js_p->error_code = 0;

    gossip_debug(GOSSIP_SETATTR_DEBUG,
                 " request has dfile_count of %d | dspace has %d\n",
                 s_op->req->u.setattr.attr.u.meta.dfile_count,
                 s_op->attr.u.meta.dfile_count);

    /* verify that the requested dfile count is sane */
    dfile_count = s_op->req->u.setattr.attr.u.meta.dfile_count;
    if ((dfile_count < 1) || (dfile_count > PVFS_REQ_LIMIT_DFILE_COUNT))
    {
        gossip_err("The requested dfile count of %d is invalid; "
                   "aborting operation.\n", dfile_count);
        js_p->error_code = -PVFS_EOVERFLOW;
        return SM_ACTION_COMPLETE;
    }

    /* set up key and value structure for keyval write */
    s_op->key.buffer = Trove_Common_Keys[METAFILE_HANDLES_KEY].key;
    s_op->key.buffer_sz = Trove_Common_Keys[METAFILE_HANDLES_KEY].size;

    gossip_debug(GOSSIP_SETATTR_DEBUG,
                 "  metafile has %d datafiles associated with it\n",
                 s_op->req->u.setattr.attr.u.meta.dfile_count);

    s_op->val.buffer = s_op->req->u.setattr.attr.u.meta.dfile_array;
    s_op->val.buffer_sz = dfile_count * sizeof(PVFS_handle);

    gossip_debug(
        GOSSIP_SETATTR_DEBUG, "  writing %s [%llu,%d,"
        "len %d]\n", (char *)s_op->key.buffer,
        llu(s_op->req->u.setattr.handle), s_op->req->u.setattr.fs_id,
        s_op->val.buffer_sz);

    /* we don't sync here since we're going to do it anyway in
     * write_metafile_distribution
     */
    ret = job_trove_keyval_write(
        s_op->req->u.setattr.fs_id, s_op->req->u.setattr.handle,
        &(s_op->key), &(s_op->val),
        0,
        NULL, smcb, 0, js_p, &j_id, server_job_context, s_op->req->hints);

    return ret;
}

static PINT_sm_action setattr_write_metafile_distribution_if_required(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = 0;
    job_id_t j_id;

    /* if we don't need to fill in the distribution, skip it */
    if (!(s_op->req->u.setattr.attr.mask & PVFS_ATTR_META_DIST))
    {
        gossip_debug(GOSSIP_SETATTR_DEBUG,
                     "skipping distribution write\n");
        js_p->error_code = 0;
        return SM_ACTION_COMPLETE;
    }

    /* set up key and value structure for keyval write */
    s_op->key.buffer = Trove_Common_Keys[METAFILE_DIST_KEY].key;
    s_op->key.buffer_sz = Trove_Common_Keys[METAFILE_DIST_KEY].size;

    gossip_debug(GOSSIP_SETATTR_DEBUG,
                 "  metafile distribution size = %d\n",
                 (int)s_op->req->u.setattr.attr.u.meta.dist_size);

    s_op->val.buffer_sz = s_op->req->u.setattr.attr.u.meta.dist_size;

    s_op->val.buffer = malloc(s_op->val.buffer_sz);
    if(!s_op->val.buffer)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    s_op->free_val = 1;
    
    PINT_dist_encode(s_op->val.buffer, 
                     s_op->req->u.setattr.attr.u.meta.dist);
    gossip_debug(
        GOSSIP_SERVER_DEBUG, "  writing %s [%llu,%d,"
        "len %d]\n", (char *)s_op->key.buffer,
        llu(s_op->req->u.setattr.handle), s_op->req->u.setattr.fs_id,
        s_op->val.buffer_sz);

    ret = job_trove_keyval_write(
        s_op->req->u.setattr.fs_id, s_op->req->u.setattr.handle,
        &(s_op->key), &(s_op->val),
        TROVE_SYNC,
        NULL, smcb, 0, js_p, &j_id, server_job_context, s_op->req->hints);

    return ret;
}

static PINT_sm_action setattr_write_symlink_target_if_required(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = 0;
    job_id_t j_id;

    /* if we don't need to fill in the symlink target, skip it */
    if (!(s_op->req->u.setattr.attr.mask & PVFS_ATTR_SYMLNK_TARGET))
    {
        gossip_debug(GOSSIP_SETATTR_DEBUG,
                     "skipping symlink target write\n");
        js_p->error_code = 0;
        return SM_ACTION_COMPLETE;
    }

    assert(s_op->req->u.setattr.attr.u.sym.target_path_len > 0);
    assert(s_op->req->u.setattr.attr.u.sym.target_path);

    /* set up key and value structure for keyval write */
    s_op->key.buffer = Trove_Common_Keys[SYMLINK_TARGET_KEY].key;
    s_op->key.buffer_sz = Trove_Common_Keys[SYMLINK_TARGET_KEY].size;

    gossip_debug(GOSSIP_SETATTR_DEBUG,
                 "  symlink target_path_len = %d\n",
                 s_op->req->u.setattr.attr.u.sym.target_path_len);

    s_op->val.buffer = s_op->req->u.setattr.attr.u.sym.target_path;
    s_op->val.buffer_sz = s_op->req->u.setattr.attr.u.sym.target_path_len;

    gossip_debug(GOSSIP_SETATTR_DEBUG, "  writing %s [%llu,%d,"
                 "len %d]\n", (char *)s_op->key.buffer,
                 llu(s_op->req->u.setattr.handle),
                 s_op->req->u.setattr.fs_id,
                 s_op->val.buffer_sz);

    ret = job_trove_keyval_write(
        s_op->req->u.setattr.fs_id, s_op->req->u.setattr.handle,
        &(s_op->key), &(s_op->val),
        TROVE_SYNC,
        NULL, smcb, 0, js_p, &j_id, server_job_context, s_op->req->hints);

    return ret;
}

/*
 * Function: setattr_cleanup
 *
 * Params:   server_op *b, 
 *           job_status_s *js_p
 *
 * Returns:  int
 *
 * Synopsis: free memory and return
 *           
 */
static PINT_sm_action setattr_cleanup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    if(s_op->free_val)
    {
        free(s_op->val.buffer);
    }
    return(server_state_machine_complete(smcb));
}

static int perm_setattr(PINT_server_op *s_op)
{
    int ret;

    if (s_op->req->capability.op_mask & PINT_CAP_SETATTR)
    {
        ret = 0;
    }
    else
    {
        ret = -PVFS_EACCES;
    }

    return ret;
}

PINT_GET_OBJECT_REF_DEFINE(setattr);
PINT_GET_CREDENTIAL_DEFINE(setattr);

struct PINT_server_req_params pvfs2_set_attr_params =
{
    .string_name = "setattr",
    .perm = perm_setattr,
    .access_type = PINT_server_req_modify,
    .sched_policy = PINT_SERVER_REQ_SCHEDULE,
    .get_object_ref = PINT_get_object_ref_setattr,
    .get_credential = PINT_get_credential_setattr,
    .state_machine = &pvfs2_set_attr_sm
};

/*
 * Local variables:
 *  mode: c
 *  c-indent-level: 4
 *  c-basic-offset: 4
 * End:
 *
 * vim: ft=c ts=8 sts=4 sw=4 expandtab
 */