lustre/ptlrpc/sec.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  *
  32  * Copyright (c) 2011, 2012, Whamcloud, Inc.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * lustre/ptlrpc/sec.c
  39  *
  40  * Author: Eric Mei <ericm@clusterfs.com>
  41  */
  42
  43 #ifndef EXPORT_SYMTAB
  44 #define EXPORT_SYMTAB
  45 #endif
  46 #define DEBUG_SUBSYSTEM S_SEC
  47
  48 #include <libcfs/libcfs.h>
  49 #ifndef __KERNEL__
  50 #include <liblustre.h>
  51 #include <libcfs/list.h>
  52 #else
  53 #include <linux/crypto.h>
  54 #include <linux/key.h>
  55 #endif
  56
  57 #include <obd.h>
  58 #include <obd_class.h>
  59 #include <obd_support.h>
  60 #include <lustre_net.h>
  61 #include <lustre_import.h>
  62 #include <lustre_dlm.h>
  63 #include <lustre_sec.h>
  64
  65 #include "ptlrpc_internal.h"
  66
  67 /***********************************************
  68  * policy registers                            *
  69  ***********************************************/
  70
  71 static cfs_rwlock_t policy_lock;
  72 static struct ptlrpc_sec_policy *policies[SPTLRPC_POLICY_MAX] = {
  73         NULL,
  74 };
  75
  76 int sptlrpc_register_policy(struct ptlrpc_sec_policy *policy)
  77 {
  78         __u16 number = policy->sp_policy;
  79
  80         LASSERT(policy->sp_name);
  81         LASSERT(policy->sp_cops);
  82         LASSERT(policy->sp_sops);
  83
  84         if (number >= SPTLRPC_POLICY_MAX)
  85                 return -EINVAL;
  86
  87         cfs_write_lock(&policy_lock);
  88         if (unlikely(policies[number])) {
  89                 cfs_write_unlock(&policy_lock);
  90                 return -EALREADY;
  91         }
  92         policies[number] = policy;
  93         cfs_write_unlock(&policy_lock);
  94
  95         CDEBUG(D_SEC, "%s: registered\n", policy->sp_name);
  96         return 0;
  97 }
  98 EXPORT_SYMBOL(sptlrpc_register_policy);
  99
 100 int sptlrpc_unregister_policy(struct ptlrpc_sec_policy *policy)
 101 {
 102         __u16 number = policy->sp_policy;
 103
 104         LASSERT(number < SPTLRPC_POLICY_MAX);
 105
 106         cfs_write_lock(&policy_lock);
 107         if (unlikely(policies[number] == NULL)) {
 108                 cfs_write_unlock(&policy_lock);
 109                 CERROR("%s: already unregistered\n", policy->sp_name);
 110                 return -EINVAL;
 111         }
 112
 113         LASSERT(policies[number] == policy);
 114         policies[number] = NULL;
 115         cfs_write_unlock(&policy_lock);
 116
 117         CDEBUG(D_SEC, "%s: unregistered\n", policy->sp_name);
 118         return 0;
 119 }
 120 EXPORT_SYMBOL(sptlrpc_unregister_policy);
 121
 122 static
 123 struct ptlrpc_sec_policy * sptlrpc_wireflavor2policy(__u32 flavor)
 124 {
 125         static CFS_DEFINE_MUTEX(load_mutex);
 126         static cfs_atomic_t       loaded = CFS_ATOMIC_INIT(0);
 127         struct ptlrpc_sec_policy *policy;
 128         __u16                     number = SPTLRPC_FLVR_POLICY(flavor);
 129         __u16                     flag = 0;
 130
 131         if (number >= SPTLRPC_POLICY_MAX)
 132                 return NULL;
 133
 134         while (1) {
 135                 cfs_read_lock(&policy_lock);
 136                 policy = policies[number];
 137                 if (policy && !cfs_try_module_get(policy->sp_owner))
 138                         policy = NULL;
 139                 if (policy == NULL)
 140                         flag = cfs_atomic_read(&loaded);
 141                 cfs_read_unlock(&policy_lock);
 142
 143                 if (policy != NULL || flag != 0 ||
 144                     number != SPTLRPC_POLICY_GSS)
 145                         break;
 146
 147                 /* try to load gss module, once */
 148                 cfs_mutex_lock(&load_mutex);
 149                 if (cfs_atomic_read(&loaded) == 0) {
 150                         if (cfs_request_module("ptlrpc_gss") == 0)
 151                                 CDEBUG(D_SEC,
 152                                        "module ptlrpc_gss loaded on demand\n");
 153                         else
 154                                 CERROR("Unable to load module ptlrpc_gss\n");
 155
 156                         cfs_atomic_set(&loaded, 1);
 157                 }
 158                 cfs_mutex_unlock(&load_mutex);
 159         }
 160
 161         return policy;
 162 }
 163
 164 __u32 sptlrpc_name2flavor_base(const char *name)
 165 {
 166         if (!strcmp(name, "null"))
 167                 return SPTLRPC_FLVR_NULL;
 168         if (!strcmp(name, "plain"))
 169                 return SPTLRPC_FLVR_PLAIN;
 170         if (!strcmp(name, "krb5n"))
 171                 return SPTLRPC_FLVR_KRB5N;
 172         if (!strcmp(name, "krb5a"))
 173                 return SPTLRPC_FLVR_KRB5A;
 174         if (!strcmp(name, "krb5i"))
 175                 return SPTLRPC_FLVR_KRB5I;
 176         if (!strcmp(name, "krb5p"))
 177                 return SPTLRPC_FLVR_KRB5P;
 178
 179         return SPTLRPC_FLVR_INVALID;
 180 }
 181 EXPORT_SYMBOL(sptlrpc_name2flavor_base);
 182
 183 const char *sptlrpc_flavor2name_base(__u32 flvr)
 184 {
 185         __u32   base = SPTLRPC_FLVR_BASE(flvr);
 186
 187         if (base == SPTLRPC_FLVR_BASE(SPTLRPC_FLVR_NULL))
 188                 return "null";
 189         else if (base == SPTLRPC_FLVR_BASE(SPTLRPC_FLVR_PLAIN))
 190                 return "plain";
 191         else if (base == SPTLRPC_FLVR_BASE(SPTLRPC_FLVR_KRB5N))
 192                 return "krb5n";
 193         else if (base == SPTLRPC_FLVR_BASE(SPTLRPC_FLVR_KRB5A))
 194                 return "krb5a";
 195         else if (base == SPTLRPC_FLVR_BASE(SPTLRPC_FLVR_KRB5I))
 196                 return "krb5i";
 197         else if (base == SPTLRPC_FLVR_BASE(SPTLRPC_FLVR_KRB5P))
 198                 return "krb5p";
 199
 200         CERROR("invalid wire flavor 0x%x\n", flvr);
 201         return "invalid";
 202 }
 203 EXPORT_SYMBOL(sptlrpc_flavor2name_base);
 204
 205 char *sptlrpc_flavor2name_bulk(struct sptlrpc_flavor *sf,
 206                                char *buf, int bufsize)
 207 {
 208         if (SPTLRPC_FLVR_POLICY(sf->sf_rpc) == SPTLRPC_POLICY_PLAIN)
 209                 snprintf(buf, bufsize, "hash:%s",
 210                          sptlrpc_get_hash_name(sf->u_bulk.hash.hash_alg));
 211         else
 212                 snprintf(buf, bufsize, "%s",
 213                          sptlrpc_flavor2name_base(sf->sf_rpc));
 214
 215         buf[bufsize - 1] = '\0';
 216         return buf;
 217 }
 218 EXPORT_SYMBOL(sptlrpc_flavor2name_bulk);
 219
 220 char *sptlrpc_flavor2name(struct sptlrpc_flavor *sf, char *buf, int bufsize)
 221 {
 222         snprintf(buf, bufsize, "%s", sptlrpc_flavor2name_base(sf->sf_rpc));
 223
 224         /*
 225          * currently we don't support customized bulk specification for
 226          * flavors other than plain
 227          */
 228         if (SPTLRPC_FLVR_POLICY(sf->sf_rpc) == SPTLRPC_POLICY_PLAIN) {
 229                 char bspec[16];
 230
 231                 bspec[0] = '-';
 232                 sptlrpc_flavor2name_bulk(sf, &bspec[1], sizeof(bspec) - 1);
 233                 strncat(buf, bspec, bufsize);
 234         }
 235
 236         buf[bufsize - 1] = '\0';
 237         return buf;
 238 }
 239 EXPORT_SYMBOL(sptlrpc_flavor2name);
 240
 241 char *sptlrpc_secflags2str(__u32 flags, char *buf, int bufsize)
 242 {
 243         buf[0] = '\0';
 244
 245         if (flags & PTLRPC_SEC_FL_REVERSE)
 246                 strncat(buf, "reverse,", bufsize);
 247         if (flags & PTLRPC_SEC_FL_ROOTONLY)
 248                 strncat(buf, "rootonly,", bufsize);
 249         if (flags & PTLRPC_SEC_FL_UDESC)
 250                 strncat(buf, "udesc,", bufsize);
 251         if (flags & PTLRPC_SEC_FL_BULK)
 252                 strncat(buf, "bulk,", bufsize);
 253         if (buf[0] == '\0')
 254                 strncat(buf, "-,", bufsize);
 255
 256         buf[bufsize - 1] = '\0';
 257         return buf;
 258 }
 259 EXPORT_SYMBOL(sptlrpc_secflags2str);
 260
 261 /**************************************************
 262  * client context APIs                            *
 263  **************************************************/
 264
 265 static
 266 struct ptlrpc_cli_ctx *get_my_ctx(struct ptlrpc_sec *sec)
 267 {
 268         struct vfs_cred vcred;
 269         int create = 1, remove_dead = 1;
 270
 271         LASSERT(sec);
 272         LASSERT(sec->ps_policy->sp_cops->lookup_ctx);
 273
 274         if (sec->ps_flvr.sf_flags & (PTLRPC_SEC_FL_REVERSE |
 275                                      PTLRPC_SEC_FL_ROOTONLY)) {
 276                 vcred.vc_uid = 0;
 277                 vcred.vc_gid = 0;
 278                 if (sec->ps_flvr.sf_flags & PTLRPC_SEC_FL_REVERSE) {
 279                         create = 0;
 280                         remove_dead = 0;
 281                 }
 282         } else {
 283                 vcred.vc_uid = cfs_curproc_uid();
 284                 vcred.vc_gid = cfs_curproc_gid();
 285         }
 286
 287         return sec->ps_policy->sp_cops->lookup_ctx(sec, &vcred,
 288                                                    create, remove_dead);
 289 }
 290
 291 struct ptlrpc_cli_ctx *sptlrpc_cli_ctx_get(struct ptlrpc_cli_ctx *ctx)
 292 {
 293         cfs_atomic_inc(&ctx->cc_refcount);
 294         return ctx;
 295 }
 296 EXPORT_SYMBOL(sptlrpc_cli_ctx_get);
 297
 298 void sptlrpc_cli_ctx_put(struct ptlrpc_cli_ctx *ctx, int sync)
 299 {
 300         struct ptlrpc_sec *sec = ctx->cc_sec;
 301
 302         LASSERT(sec);
 303         LASSERT_ATOMIC_POS(&ctx->cc_refcount);
 304
 305         if (!cfs_atomic_dec_and_test(&ctx->cc_refcount))
 306                 return;
 307
 308         sec->ps_policy->sp_cops->release_ctx(sec, ctx, sync);
 309 }
 310 EXPORT_SYMBOL(sptlrpc_cli_ctx_put);
 311
 312 /**
 313  * Expire the client context immediately.
 314  *
 315  * \pre Caller must hold at least 1 reference on the \a ctx.
 316  */
 317 void sptlrpc_cli_ctx_expire(struct ptlrpc_cli_ctx *ctx)
 318 {
 319         LASSERT(ctx->cc_ops->die);
 320         ctx->cc_ops->die(ctx, 0);
 321 }
 322 EXPORT_SYMBOL(sptlrpc_cli_ctx_expire);
 323
 324 /**
 325  * To wake up the threads who are waiting for this client context. Called
 326  * after some status change happened on \a ctx.
 327  */
 328 void sptlrpc_cli_ctx_wakeup(struct ptlrpc_cli_ctx *ctx)
 329 {
 330         struct ptlrpc_request *req, *next;
 331
 332         cfs_spin_lock(&ctx->cc_lock);
 333         cfs_list_for_each_entry_safe(req, next, &ctx->cc_req_list,
 334                                      rq_ctx_chain) {
 335                 cfs_list_del_init(&req->rq_ctx_chain);
 336                 ptlrpc_client_wake_req(req);
 337         }
 338         cfs_spin_unlock(&ctx->cc_lock);
 339 }
 340 EXPORT_SYMBOL(sptlrpc_cli_ctx_wakeup);
 341
 342 int sptlrpc_cli_ctx_display(struct ptlrpc_cli_ctx *ctx, char *buf, int bufsize)
 343 {
 344         LASSERT(ctx->cc_ops);
 345
 346         if (ctx->cc_ops->display == NULL)
 347                 return 0;
 348
 349         return ctx->cc_ops->display(ctx, buf, bufsize);
 350 }
 351
 352 static int import_sec_check_expire(struct obd_import *imp)
 353 {
 354         int     adapt = 0;
 355
 356         cfs_spin_lock(&imp->imp_lock);
 357         if (imp->imp_sec_expire &&
 358             imp->imp_sec_expire < cfs_time_current_sec()) {
 359                 adapt = 1;
 360                 imp->imp_sec_expire = 0;
 361         }
 362         cfs_spin_unlock(&imp->imp_lock);
 363
 364         if (!adapt)
 365                 return 0;
 366
 367         CDEBUG(D_SEC, "found delayed sec adapt expired, do it now\n");
 368         return sptlrpc_import_sec_adapt(imp, NULL, 0);
 369 }
 370
 371 static int import_sec_validate_get(struct obd_import *imp,
 372                                    struct ptlrpc_sec **sec)
 373 {
 374         int     rc;
 375
 376         if (unlikely(imp->imp_sec_expire)) {
 377                 rc = import_sec_check_expire(imp);
 378                 if (rc)
 379                         return rc;
 380         }
 381
 382         *sec = sptlrpc_import_sec_ref(imp);
 383         if (*sec == NULL) {
 384                 CERROR("import %p (%s) with no sec\n",
 385                        imp, ptlrpc_import_state_name(imp->imp_state));
 386                 return -EACCES;
 387         }
 388
 389         if (unlikely((*sec)->ps_dying)) {
 390                 CERROR("attempt to use dying sec %p\n", sec);
 391                 sptlrpc_sec_put(*sec);
 392                 return -EACCES;
 393         }
 394
 395         return 0;
 396 }
 397
 398 /**
 399  * Given a \a req, find or allocate a appropriate context for it.
 400  * \pre req->rq_cli_ctx == NULL.
 401  *
 402  * \retval 0 succeed, and req->rq_cli_ctx is set.
 403  * \retval -ev error number, and req->rq_cli_ctx == NULL.
 404  */
 405 int sptlrpc_req_get_ctx(struct ptlrpc_request *req)
 406 {
 407         struct obd_import *imp = req->rq_import;
 408         struct ptlrpc_sec *sec;
 409         int                rc;
 410         ENTRY;
 411
 412         LASSERT(!req->rq_cli_ctx);
 413         LASSERT(imp);
 414
 415         rc = import_sec_validate_get(imp, &sec);
 416         if (rc)
 417                 RETURN(rc);
 418
 419         req->rq_cli_ctx = get_my_ctx(sec);
 420
 421         sptlrpc_sec_put(sec);
 422
 423         if (!req->rq_cli_ctx) {
 424                 CERROR("req %p: fail to get context\n", req);
 425                 RETURN(-ENOMEM);
 426         }
 427
 428         RETURN(0);
 429 }
 430
 431 /**
 432  * Drop the context for \a req.
 433  * \pre req->rq_cli_ctx != NULL.
 434  * \post req->rq_cli_ctx == NULL.
 435  *
 436  * If \a sync == 0, this function should return quickly without sleep;
 437  * otherwise it might trigger and wait for the whole process of sending
 438  * an context-destroying rpc to server.
 439  */
 440 void sptlrpc_req_put_ctx(struct ptlrpc_request *req, int sync)
 441 {
 442         ENTRY;
 443
 444         LASSERT(req);
 445         LASSERT(req->rq_cli_ctx);
 446
 447         /* request might be asked to release earlier while still
 448          * in the context waiting list.
 449          */
 450         if (!cfs_list_empty(&req->rq_ctx_chain)) {
 451                 cfs_spin_lock(&req->rq_cli_ctx->cc_lock);
 452                 cfs_list_del_init(&req->rq_ctx_chain);
 453                 cfs_spin_unlock(&req->rq_cli_ctx->cc_lock);
 454         }
 455
 456         sptlrpc_cli_ctx_put(req->rq_cli_ctx, sync);
 457         req->rq_cli_ctx = NULL;
 458         EXIT;
 459 }
 460
 461 static
 462 int sptlrpc_req_ctx_switch(struct ptlrpc_request *req,
 463                            struct ptlrpc_cli_ctx *oldctx,
 464                            struct ptlrpc_cli_ctx *newctx)
 465 {
 466         struct sptlrpc_flavor   old_flvr;
 467         char                   *reqmsg = NULL; /* to workaround old gcc */
 468         int                     reqmsg_size;
 469         int                     rc = 0;
 470
 471         LASSERT(req->rq_reqmsg);
 472         LASSERT(req->rq_reqlen);
 473         LASSERT(req->rq_replen);
 474
 475         CDEBUG(D_SEC, "req %p: switch ctx %p(%u->%s) -> %p(%u->%s), "
 476                "switch sec %p(%s) -> %p(%s)\n", req,
 477                oldctx, oldctx->cc_vcred.vc_uid, sec2target_str(oldctx->cc_sec),
 478                newctx, newctx->cc_vcred.vc_uid, sec2target_str(newctx->cc_sec),
 479                oldctx->cc_sec, oldctx->cc_sec->ps_policy->sp_name,
 480                newctx->cc_sec, newctx->cc_sec->ps_policy->sp_name);
 481
 482         /* save flavor */
 483         old_flvr = req->rq_flvr;
 484
 485         /* save request message */
 486         reqmsg_size = req->rq_reqlen;
 487         if (reqmsg_size != 0) {
 488                 OBD_ALLOC_LARGE(reqmsg, reqmsg_size);
 489                 if (reqmsg == NULL)
 490                         return -ENOMEM;
 491                 memcpy(reqmsg, req->rq_reqmsg, reqmsg_size);
 492         }
 493
 494         /* release old req/rep buf */
 495         req->rq_cli_ctx = oldctx;
 496         sptlrpc_cli_free_reqbuf(req);
 497         sptlrpc_cli_free_repbuf(req);
 498         req->rq_cli_ctx = newctx;
 499
 500         /* recalculate the flavor */
 501         sptlrpc_req_set_flavor(req, 0);
 502
 503         /* alloc new request buffer
 504          * we don't need to alloc reply buffer here, leave it to the
 505          * rest procedure of ptlrpc */
 506         if (reqmsg_size != 0) {
 507                 rc = sptlrpc_cli_alloc_reqbuf(req, reqmsg_size);
 508                 if (!rc) {
 509                         LASSERT(req->rq_reqmsg);
 510                         memcpy(req->rq_reqmsg, reqmsg, reqmsg_size);
 511                 } else {
 512                         CWARN("failed to alloc reqbuf: %d\n", rc);
 513                         req->rq_flvr = old_flvr;
 514                 }
 515
 516                 OBD_FREE_LARGE(reqmsg, reqmsg_size);
 517         }
 518         return rc;
 519 }
 520
 521 /**
 522  * If current context of \a req is dead somehow, e.g. we just switched flavor
 523  * thus marked original contexts dead, we'll find a new context for it. if
 524  * no switch is needed, \a req will end up with the same context.
 525  *
 526  * \note a request must have a context, to keep other parts of code happy.
 527  * In any case of failure during the switching, we must restore the old one.
 528  */
 529 int sptlrpc_req_replace_dead_ctx(struct ptlrpc_request *req)
 530 {
 531         struct ptlrpc_cli_ctx *oldctx = req->rq_cli_ctx;
 532         struct ptlrpc_cli_ctx *newctx;
 533         int                    rc;
 534         ENTRY;
 535
 536         LASSERT(oldctx);
 537
 538         sptlrpc_cli_ctx_get(oldctx);
 539         sptlrpc_req_put_ctx(req, 0);
 540
 541         rc = sptlrpc_req_get_ctx(req);
 542         if (unlikely(rc)) {
 543                 LASSERT(!req->rq_cli_ctx);
 544
 545                 /* restore old ctx */
 546                 req->rq_cli_ctx = oldctx;
 547                 RETURN(rc);
 548         }
 549
 550         newctx = req->rq_cli_ctx;
 551         LASSERT(newctx);
 552
 553         if (unlikely(newctx == oldctx &&
 554                      cfs_test_bit(PTLRPC_CTX_DEAD_BIT, &oldctx->cc_flags))) {
 555                 /*
 556                  * still get the old dead ctx, usually means system too busy
 557                  */
 558                 CDEBUG(D_SEC,
 559                        "ctx (%p, fl %lx) doesn't switch, relax a little bit\n",
 560                        newctx, newctx->cc_flags);
 561
 562                 cfs_schedule_timeout_and_set_state(CFS_TASK_INTERRUPTIBLE,
 563                                                    CFS_HZ);
 564         } else {
 565                 /*
 566                  * it's possible newctx == oldctx if we're switching
 567                  * subflavor with the same sec.
 568                  */
 569                 rc = sptlrpc_req_ctx_switch(req, oldctx, newctx);
 570                 if (rc) {
 571                         /* restore old ctx */
 572                         sptlrpc_req_put_ctx(req, 0);
 573                         req->rq_cli_ctx = oldctx;
 574                         RETURN(rc);
 575                 }
 576
 577                 LASSERT(req->rq_cli_ctx == newctx);
 578         }
 579
 580         sptlrpc_cli_ctx_put(oldctx, 1);
 581         RETURN(0);
 582 }
 583 EXPORT_SYMBOL(sptlrpc_req_replace_dead_ctx);
 584
 585 static
 586 int ctx_check_refresh(struct ptlrpc_cli_ctx *ctx)
 587 {
 588         if (cli_ctx_is_refreshed(ctx))
 589                 return 1;
 590         return 0;
 591 }
 592
 593 static
 594 int ctx_refresh_timeout(void *data)
 595 {
 596         struct ptlrpc_request *req = data;
 597         int rc;
 598
 599         /* conn_cnt is needed in expire_one_request */
 600         lustre_msg_set_conn_cnt(req->rq_reqmsg, req->rq_import->imp_conn_cnt);
 601
 602         rc = ptlrpc_expire_one_request(req, 1);
 603         /* if we started recovery, we should mark this ctx dead; otherwise
 604          * in case of lgssd died nobody would retire this ctx, following
 605          * connecting will still find the same ctx thus cause deadlock.
 606          * there's an assumption that expire time of the request should be
 607          * later than the context refresh expire time.
 608          */
 609         if (rc == 0)
 610                 req->rq_cli_ctx->cc_ops->die(req->rq_cli_ctx, 0);
 611         return rc;
 612 }
 613
 614 static
 615 void ctx_refresh_interrupt(void *data)
 616 {
 617         struct ptlrpc_request *req = data;
 618
 619         cfs_spin_lock(&req->rq_lock);
 620         req->rq_intr = 1;
 621         cfs_spin_unlock(&req->rq_lock);
 622 }
 623
 624 static
 625 void req_off_ctx_list(struct ptlrpc_request *req, struct ptlrpc_cli_ctx *ctx)
 626 {
 627         cfs_spin_lock(&ctx->cc_lock);
 628         if (!cfs_list_empty(&req->rq_ctx_chain))
 629                 cfs_list_del_init(&req->rq_ctx_chain);
 630         cfs_spin_unlock(&ctx->cc_lock);
 631 }
 632
 633 /**
 634  * To refresh the context of \req, if it's not up-to-date.
 635  * \param timeout
 636  * - < 0: don't wait
 637  * - = 0: wait until success or fatal error occur
 638  * - > 0: timeout value (in seconds)
 639  *
 640  * The status of the context could be subject to be changed by other threads
 641  * at any time. We allow this race, but once we return with 0, the caller will
 642  * suppose it's uptodated and keep using it until the owning rpc is done.
 643  *
 644  * \retval 0 only if the context is uptodated.
 645  * \retval -ev error number.
 646  */
 647 int sptlrpc_req_refresh_ctx(struct ptlrpc_request *req, long timeout)
 648 {
 649         struct ptlrpc_cli_ctx  *ctx = req->rq_cli_ctx;
 650         struct ptlrpc_sec      *sec;
 651         struct l_wait_info      lwi;
 652         int                     rc;
 653         ENTRY;
 654
 655         LASSERT(ctx);
 656
 657         if (req->rq_ctx_init || req->rq_ctx_fini)
 658                 RETURN(0);
 659
 660         /*
 661          * during the process a request's context might change type even
 662          * (e.g. from gss ctx to null ctx), so each loop we need to re-check
 663          * everything
 664          */
 665 again:
 666         rc = import_sec_validate_get(req->rq_import, &sec);
 667         if (rc)
 668                 RETURN(rc);
 669
 670         if (sec->ps_flvr.sf_rpc != req->rq_flvr.sf_rpc) {
 671                 CDEBUG(D_SEC, "req %p: flavor has changed %x -> %x\n",
 672                       req, req->rq_flvr.sf_rpc, sec->ps_flvr.sf_rpc);
 673                 req_off_ctx_list(req, ctx);
 674                 sptlrpc_req_replace_dead_ctx(req);
 675                 ctx = req->rq_cli_ctx;
 676         }
 677         sptlrpc_sec_put(sec);
 678
 679         if (cli_ctx_is_eternal(ctx))
 680                 RETURN(0);
 681
 682         if (unlikely(cfs_test_bit(PTLRPC_CTX_NEW_BIT, &ctx->cc_flags))) {
 683                 LASSERT(ctx->cc_ops->refresh);
 684                 ctx->cc_ops->refresh(ctx);
 685         }
 686         LASSERT(cfs_test_bit(PTLRPC_CTX_NEW_BIT, &ctx->cc_flags) == 0);
 687
 688         LASSERT(ctx->cc_ops->validate);
 689         if (ctx->cc_ops->validate(ctx) == 0) {
 690                 req_off_ctx_list(req, ctx);
 691                 RETURN(0);
 692         }
 693
 694         if (unlikely(cfs_test_bit(PTLRPC_CTX_ERROR_BIT, &ctx->cc_flags))) {
 695                 cfs_spin_lock(&req->rq_lock);
 696                 req->rq_err = 1;
 697                 cfs_spin_unlock(&req->rq_lock);
 698                 req_off_ctx_list(req, ctx);
 699                 RETURN(-EPERM);
 700         }
 701
 702         /*
 703          * There's a subtle issue for resending RPCs, suppose following
 704          * situation:
 705          *  1. the request was sent to server.
 706          *  2. recovery was kicked start, after finished the request was
 707          *     marked as resent.
 708          *  3. resend the request.
 709          *  4. old reply from server received, we accept and verify the reply.
 710          *     this has to be success, otherwise the error will be aware
 711          *     by application.
 712          *  5. new reply from server received, dropped by LNet.
 713          *
 714          * Note the xid of old & new request is the same. We can't simply
 715          * change xid for the resent request because the server replies on
 716          * it for reply reconstruction.
 717          *
 718          * Commonly the original context should be uptodate because we
 719          * have a expiry nice time; server will keep its context because
 720          * we at least hold a ref of old context which prevent context
 721          * destroying RPC being sent. So server still can accept the request
 722          * and finish the RPC. But if that's not the case:
 723          *  1. If server side context has been trimmed, a NO_CONTEXT will
 724          *     be returned, gss_cli_ctx_verify/unseal will switch to new
 725          *     context by force.
 726          *  2. Current context never be refreshed, then we are fine: we
 727          *     never really send request with old context before.
 728          */
 729         if (cfs_test_bit(PTLRPC_CTX_UPTODATE_BIT, &ctx->cc_flags) &&
 730             unlikely(req->rq_reqmsg) &&
 731             lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT) {
 732                 req_off_ctx_list(req, ctx);
 733                 RETURN(0);
 734         }
 735
 736         if (unlikely(cfs_test_bit(PTLRPC_CTX_DEAD_BIT, &ctx->cc_flags))) {
 737                 req_off_ctx_list(req, ctx);
 738                 /*
 739                  * don't switch ctx if import was deactivated
 740                  */
 741                 if (req->rq_import->imp_deactive) {
 742                         cfs_spin_lock(&req->rq_lock);
 743                         req->rq_err = 1;
 744                         cfs_spin_unlock(&req->rq_lock);
 745                         RETURN(-EINTR);
 746                 }
 747
 748                 rc = sptlrpc_req_replace_dead_ctx(req);
 749                 if (rc) {
 750                         LASSERT(ctx == req->rq_cli_ctx);
 751                         CERROR("req %p: failed to replace dead ctx %p: %d\n",
 752                                 req, ctx, rc);
 753                         cfs_spin_lock(&req->rq_lock);
 754                         req->rq_err = 1;
 755                         cfs_spin_unlock(&req->rq_lock);
 756                         RETURN(rc);
 757                 }
 758
 759                 ctx = req->rq_cli_ctx;
 760                 goto again;
 761         }
 762
 763         /*
 764          * Now we're sure this context is during upcall, add myself into
 765          * waiting list
 766          */
 767         cfs_spin_lock(&ctx->cc_lock);
 768         if (cfs_list_empty(&req->rq_ctx_chain))
 769                 cfs_list_add(&req->rq_ctx_chain, &ctx->cc_req_list);
 770         cfs_spin_unlock(&ctx->cc_lock);
 771
 772         if (timeout < 0)
 773                 RETURN(-EWOULDBLOCK);
 774
 775         /* Clear any flags that may be present from previous sends */
 776         LASSERT(req->rq_receiving_reply == 0);
 777         cfs_spin_lock(&req->rq_lock);
 778         req->rq_err = 0;
 779         req->rq_timedout = 0;
 780         req->rq_resend = 0;
 781         req->rq_restart = 0;
 782         cfs_spin_unlock(&req->rq_lock);
 783
 784         lwi = LWI_TIMEOUT_INTR(timeout * CFS_HZ, ctx_refresh_timeout,
 785                                ctx_refresh_interrupt, req);
 786         rc = l_wait_event(req->rq_reply_waitq, ctx_check_refresh(ctx), &lwi);
 787
 788         /*
 789          * following cases could lead us here:
 790          * - successfully refreshed;
 791          * - interrupted;
 792          * - timedout, and we don't want recover from the failure;
 793          * - timedout, and waked up upon recovery finished;
 794          * - someone else mark this ctx dead by force;
 795          * - someone invalidate the req and call ptlrpc_client_wake_req(),
 796          *   e.g. ptlrpc_abort_inflight();
 797          */
 798         if (!cli_ctx_is_refreshed(ctx)) {
 799                 /* timed out or interruptted */
 800                 req_off_ctx_list(req, ctx);
 801
 802                 LASSERT(rc != 0);
 803                 RETURN(rc);
 804         }
 805
 806         goto again;
 807 }
 808
 809 /**
 810  * Initialize flavor settings for \a req, according to \a opcode.
 811  *
 812  * \note this could be called in two situations:
 813  * - new request from ptlrpc_pre_req(), with proper @opcode
 814  * - old request which changed ctx in the middle, with @opcode == 0
 815  */
 816 void sptlrpc_req_set_flavor(struct ptlrpc_request *req, int opcode)
 817 {
 818         struct ptlrpc_sec *sec;
 819
 820         LASSERT(req->rq_import);
 821         LASSERT(req->rq_cli_ctx);
 822         LASSERT(req->rq_cli_ctx->cc_sec);
 823         LASSERT(req->rq_bulk_read == 0 || req->rq_bulk_write == 0);
 824
 825         /* special security flags accoding to opcode */
 826         switch (opcode) {
 827         case OST_READ:
 828         case MDS_READPAGE:
 829         case MGS_CONFIG_READ:
 830                 req->rq_bulk_read = 1;
 831                 break;
 832         case OST_WRITE:
 833         case MDS_WRITEPAGE:
 834                 req->rq_bulk_write = 1;
 835                 break;
 836         case SEC_CTX_INIT:
 837                 req->rq_ctx_init = 1;
 838                 break;
 839         case SEC_CTX_FINI:
 840                 req->rq_ctx_fini = 1;
 841                 break;
 842         case 0:
 843                 /* init/fini rpc won't be resend, so can't be here */
 844                 LASSERT(req->rq_ctx_init == 0);
 845                 LASSERT(req->rq_ctx_fini == 0);
 846
 847                 /* cleanup flags, which should be recalculated */
 848                 req->rq_pack_udesc = 0;
 849                 req->rq_pack_bulk = 0;
 850                 break;
 851         }
 852
 853         sec = req->rq_cli_ctx->cc_sec;
 854
 855         cfs_spin_lock(&sec->ps_lock);
 856         req->rq_flvr = sec->ps_flvr;
 857         cfs_spin_unlock(&sec->ps_lock);
 858
 859         /* force SVC_NULL for context initiation rpc, SVC_INTG for context
 860          * destruction rpc */
 861         if (unlikely(req->rq_ctx_init))
 862                 flvr_set_svc(&req->rq_flvr.sf_rpc, SPTLRPC_SVC_NULL);
 863         else if (unlikely(req->rq_ctx_fini))
 864                 flvr_set_svc(&req->rq_flvr.sf_rpc, SPTLRPC_SVC_INTG);
 865
 866         /* user descriptor flag, null security can't do it anyway */
 867         if ((sec->ps_flvr.sf_flags & PTLRPC_SEC_FL_UDESC) &&
 868             (req->rq_flvr.sf_rpc != SPTLRPC_FLVR_NULL))
 869                 req->rq_pack_udesc = 1;
 870
 871         /* bulk security flag */
 872         if ((req->rq_bulk_read || req->rq_bulk_write) &&
 873             sptlrpc_flavor_has_bulk(&req->rq_flvr))
 874                 req->rq_pack_bulk = 1;
 875 }
 876
 877 void sptlrpc_request_out_callback(struct ptlrpc_request *req)
 878 {
 879         if (SPTLRPC_FLVR_SVC(req->rq_flvr.sf_rpc) != SPTLRPC_SVC_PRIV)
 880                 return;
 881
 882         LASSERT(req->rq_clrbuf);
 883         if (req->rq_pool || !req->rq_reqbuf)
 884                 return;
 885
 886         OBD_FREE(req->rq_reqbuf, req->rq_reqbuf_len);
 887         req->rq_reqbuf = NULL;
 888         req->rq_reqbuf_len = 0;
 889 }
 890
 891 /**
 892  * Given an import \a imp, check whether current user has a valid context
 893  * or not. We may create a new context and try to refresh it, and try
 894  * repeatedly try in case of non-fatal errors. Return 0 means success.
 895  */
 896 int sptlrpc_import_check_ctx(struct obd_import *imp)
 897 {
 898         struct ptlrpc_sec     *sec;
 899         struct ptlrpc_cli_ctx *ctx;
 900         struct ptlrpc_request *req = NULL;
 901         int rc;
 902         ENTRY;
 903
 904         cfs_might_sleep();
 905
 906         sec = sptlrpc_import_sec_ref(imp);
 907         ctx = get_my_ctx(sec);
 908         sptlrpc_sec_put(sec);
 909
 910         if (!ctx)
 911                 RETURN(-ENOMEM);
 912
 913         if (cli_ctx_is_eternal(ctx) ||
 914             ctx->cc_ops->validate(ctx) == 0) {
 915                 sptlrpc_cli_ctx_put(ctx, 1);
 916                 RETURN(0);
 917         }
 918
 919         if (cli_ctx_is_error(ctx)) {
 920                 sptlrpc_cli_ctx_put(ctx, 1);
 921                 RETURN(-EACCES);
 922         }
 923
 924         OBD_ALLOC_PTR(req);
 925         if (!req)
 926                 RETURN(-ENOMEM);
 927
 928         cfs_spin_lock_init(&req->rq_lock);
 929         cfs_atomic_set(&req->rq_refcount, 10000);
 930         CFS_INIT_LIST_HEAD(&req->rq_ctx_chain);
 931         cfs_waitq_init(&req->rq_reply_waitq);
 932         cfs_waitq_init(&req->rq_set_waitq);
 933         req->rq_import = imp;
 934         req->rq_flvr = sec->ps_flvr;
 935         req->rq_cli_ctx = ctx;
 936
 937         rc = sptlrpc_req_refresh_ctx(req, 0);
 938         LASSERT(cfs_list_empty(&req->rq_ctx_chain));
 939         sptlrpc_cli_ctx_put(req->rq_cli_ctx, 1);
 940         OBD_FREE_PTR(req);
 941
 942         RETURN(rc);
 943 }
 944
 945 /**
 946  * Used by ptlrpc client, to perform the pre-defined security transformation
 947  * upon the request message of \a req. After this function called,
 948  * req->rq_reqmsg is still accessible as clear text.
 949  */
 950 int sptlrpc_cli_wrap_request(struct ptlrpc_request *req)
 951 {
 952         struct ptlrpc_cli_ctx *ctx = req->rq_cli_ctx;
 953         int rc = 0;
 954         ENTRY;
 955
 956         LASSERT(ctx);
 957         LASSERT(ctx->cc_sec);
 958         LASSERT(req->rq_reqbuf || req->rq_clrbuf);
 959
 960         /* we wrap bulk request here because now we can be sure
 961          * the context is uptodate.
 962          */
 963         if (req->rq_bulk) {
 964                 rc = sptlrpc_cli_wrap_bulk(req, req->rq_bulk);
 965                 if (rc)
 966                         RETURN(rc);
 967         }
 968
 969         switch (SPTLRPC_FLVR_SVC(req->rq_flvr.sf_rpc)) {
 970         case SPTLRPC_SVC_NULL:
 971         case SPTLRPC_SVC_AUTH:
 972         case SPTLRPC_SVC_INTG:
 973                 LASSERT(ctx->cc_ops->sign);
 974                 rc = ctx->cc_ops->sign(ctx, req);
 975                 break;
 976         case SPTLRPC_SVC_PRIV:
 977                 LASSERT(ctx->cc_ops->seal);
 978                 rc = ctx->cc_ops->seal(ctx, req);
 979                 break;
 980         default:
 981                 LBUG();
 982         }
 983
 984         if (rc == 0) {
 985                 LASSERT(req->rq_reqdata_len);
 986                 LASSERT(req->rq_reqdata_len % 8 == 0);
 987                 LASSERT(req->rq_reqdata_len <= req->rq_reqbuf_len);
 988         }
 989
 990         RETURN(rc);
 991 }
 992
 993 static int do_cli_unwrap_reply(struct ptlrpc_request *req)
 994 {
 995         struct ptlrpc_cli_ctx *ctx = req->rq_cli_ctx;
 996         int                    rc;
 997         ENTRY;
 998
 999         LASSERT(ctx);
1000         LASSERT(ctx->cc_sec);
1001         LASSERT(req->rq_repbuf);
1002         LASSERT(req->rq_repdata);
1003         LASSERT(req->rq_repmsg == NULL);
1004
1005         req->rq_rep_swab_mask = 0;
1006
1007         rc = __lustre_unpack_msg(req->rq_repdata, req->rq_repdata_len);
1008         switch (rc) {
1009         case 1:
1010                 lustre_set_rep_swabbed(req, MSG_PTLRPC_HEADER_OFF);
1011         case 0:
1012                 break;
1013         default:
1014                 CERROR("failed unpack reply: x"LPU64"\n", req->rq_xid);
1015                 RETURN(-EPROTO);
1016         }
1017
1018         if (req->rq_repdata_len < sizeof(struct lustre_msg)) {
1019                 CERROR("replied data length %d too small\n",
1020                        req->rq_repdata_len);
1021                 RETURN(-EPROTO);
1022         }
1023
1024         if (SPTLRPC_FLVR_POLICY(req->rq_repdata->lm_secflvr) !=
1025             SPTLRPC_FLVR_POLICY(req->rq_flvr.sf_rpc)) {
1026                 CERROR("reply policy %u doesn't match request policy %u\n",
1027                        SPTLRPC_FLVR_POLICY(req->rq_repdata->lm_secflvr),
1028                        SPTLRPC_FLVR_POLICY(req->rq_flvr.sf_rpc));
1029                 RETURN(-EPROTO);
1030         }
1031
1032         switch (SPTLRPC_FLVR_SVC(req->rq_flvr.sf_rpc)) {
1033         case SPTLRPC_SVC_NULL:
1034         case SPTLRPC_SVC_AUTH:
1035         case SPTLRPC_SVC_INTG:
1036                 LASSERT(ctx->cc_ops->verify);
1037                 rc = ctx->cc_ops->verify(ctx, req);
1038                 break;
1039         case SPTLRPC_SVC_PRIV:
1040                 LASSERT(ctx->cc_ops->unseal);
1041                 rc = ctx->cc_ops->unseal(ctx, req);
1042                 break;
1043         default:
1044                 LBUG();
1045         }
1046         LASSERT(rc || req->rq_repmsg || req->rq_resend);
1047
1048         if (SPTLRPC_FLVR_POLICY(req->rq_flvr.sf_rpc) != SPTLRPC_POLICY_NULL &&
1049             !req->rq_ctx_init)
1050                 req->rq_rep_swab_mask = 0;
1051         RETURN(rc);
1052 }
1053
1054 /**
1055  * Used by ptlrpc client, to perform security transformation upon the reply
1056  * message of \a req. After return successfully, req->rq_repmsg points to
1057  * the reply message in clear text.
1058  *
1059  * \pre the reply buffer should have been un-posted from LNet, so nothing is
1060  * going to change.
1061  */
1062 int sptlrpc_cli_unwrap_reply(struct ptlrpc_request *req)
1063 {
1064         LASSERT(req->rq_repbuf);
1065         LASSERT(req->rq_repdata == NULL);
1066         LASSERT(req->rq_repmsg == NULL);
1067         LASSERT(req->rq_reply_off + req->rq_nob_received <= req->rq_repbuf_len);
1068
1069         if (req->rq_reply_off == 0 &&
1070             (lustre_msghdr_get_flags(req->rq_reqmsg) & MSGHDR_AT_SUPPORT)) {
1071                 CERROR("real reply with offset 0\n");
1072                 return -EPROTO;
1073         }
1074
1075         if (req->rq_reply_off % 8 != 0) {
1076                 CERROR("reply at odd offset %u\n", req->rq_reply_off);
1077                 return -EPROTO;
1078         }
1079
1080         req->rq_repdata = (struct lustre_msg *)
1081                                 (req->rq_repbuf + req->rq_reply_off);
1082         req->rq_repdata_len = req->rq_nob_received;
1083
1084         return do_cli_unwrap_reply(req);
1085 }
1086
1087 /**
1088  * Used by ptlrpc client, to perform security transformation upon the early
1089  * reply message of \a req. We expect the rq_reply_off is 0, and
1090  * rq_nob_received is the early reply size.
1091  *
1092  * Because the receive buffer might be still posted, the reply data might be
1093  * changed at any time, no matter we're holding rq_lock or not. For this reason
1094  * we allocate a separate ptlrpc_request and reply buffer for early reply
1095  * processing.
1096  *
1097  * \retval 0 success, \a req_ret is filled with a duplicated ptlrpc_request.
1098  * Later the caller must call sptlrpc_cli_finish_early_reply() on the returned
1099  * \a *req_ret to release it.
1100  * \retval -ev error number, and \a req_ret will not be set.
1101  */
1102 int sptlrpc_cli_unwrap_early_reply(struct ptlrpc_request *req,
1103                                    struct ptlrpc_request **req_ret)
1104 {
1105         struct ptlrpc_request  *early_req;
1106         char                   *early_buf;
1107         int                     early_bufsz, early_size;
1108         int                     rc;
1109         ENTRY;
1110
1111         OBD_ALLOC_PTR(early_req);
1112         if (early_req == NULL)
1113                 RETURN(-ENOMEM);
1114
1115         early_size = req->rq_nob_received;
1116         early_bufsz = size_roundup_power2(early_size);
1117         OBD_ALLOC_LARGE(early_buf, early_bufsz);
1118         if (early_buf == NULL)
1119                 GOTO(err_req, rc = -ENOMEM);
1120
1121         /* sanity checkings and copy data out, do it inside spinlock */
1122         cfs_spin_lock(&req->rq_lock);
1123
1124         if (req->rq_replied) {
1125                 cfs_spin_unlock(&req->rq_lock);
1126                 GOTO(err_buf, rc = -EALREADY);
1127         }
1128
1129         LASSERT(req->rq_repbuf);
1130         LASSERT(req->rq_repdata == NULL);
1131         LASSERT(req->rq_repmsg == NULL);
1132
1133         if (req->rq_reply_off != 0) {
1134                 CERROR("early reply with offset %u\n", req->rq_reply_off);
1135                 cfs_spin_unlock(&req->rq_lock);
1136                 GOTO(err_buf, rc = -EPROTO);
1137         }
1138
1139         if (req->rq_nob_received != early_size) {
1140                 /* even another early arrived the size should be the same */
1141                 CERROR("data size has changed from %u to %u\n",
1142                        early_size, req->rq_nob_received);
1143                 cfs_spin_unlock(&req->rq_lock);
1144                 GOTO(err_buf, rc = -EINVAL);
1145         }
1146
1147         if (req->rq_nob_received < sizeof(struct lustre_msg)) {
1148                 CERROR("early reply length %d too small\n",
1149                        req->rq_nob_received);
1150                 cfs_spin_unlock(&req->rq_lock);
1151                 GOTO(err_buf, rc = -EALREADY);
1152         }
1153
1154         memcpy(early_buf, req->rq_repbuf, early_size);
1155         cfs_spin_unlock(&req->rq_lock);
1156
1157         cfs_spin_lock_init(&early_req->rq_lock);
1158         early_req->rq_cli_ctx = sptlrpc_cli_ctx_get(req->rq_cli_ctx);
1159         early_req->rq_flvr = req->rq_flvr;
1160         early_req->rq_repbuf = early_buf;
1161         early_req->rq_repbuf_len = early_bufsz;
1162         early_req->rq_repdata = (struct lustre_msg *) early_buf;
1163         early_req->rq_repdata_len = early_size;
1164         early_req->rq_early = 1;
1165         early_req->rq_reqmsg = req->rq_reqmsg;
1166
1167         rc = do_cli_unwrap_reply(early_req);
1168         if (rc) {
1169                 DEBUG_REQ(D_ADAPTTO, early_req,
1170                           "error %d unwrap early reply", rc);
1171                 GOTO(err_ctx, rc);
1172         }
1173
1174         LASSERT(early_req->rq_repmsg);
1175         *req_ret = early_req;
1176         RETURN(0);
1177
1178 err_ctx:
1179         sptlrpc_cli_ctx_put(early_req->rq_cli_ctx, 1);
1180 err_buf:
1181         OBD_FREE_LARGE(early_buf, early_bufsz);
1182 err_req:
1183         OBD_FREE_PTR(early_req);
1184         RETURN(rc);
1185 }
1186
1187 /**
1188  * Used by ptlrpc client, to release a processed early reply \a early_req.
1189  *
1190  * \pre \a early_req was obtained from calling sptlrpc_cli_unwrap_early_reply().
1191  */
1192 void sptlrpc_cli_finish_early_reply(struct ptlrpc_request *early_req)
1193 {
1194         LASSERT(early_req->rq_repbuf);
1195         LASSERT(early_req->rq_repdata);
1196         LASSERT(early_req->rq_repmsg);
1197
1198         sptlrpc_cli_ctx_put(early_req->rq_cli_ctx, 1);
1199         OBD_FREE_LARGE(early_req->rq_repbuf, early_req->rq_repbuf_len);
1200         OBD_FREE_PTR(early_req);
1201 }
1202
1203 /**************************************************
1204  * sec ID                                         *
1205  **************************************************/
1206
1207 /*
1208  * "fixed" sec (e.g. null) use sec_id < 0
1209  */
1210 static cfs_atomic_t sptlrpc_sec_id = CFS_ATOMIC_INIT(1);
1211
1212 int sptlrpc_get_next_secid(void)
1213 {
1214         return cfs_atomic_inc_return(&sptlrpc_sec_id);
1215 }
1216 EXPORT_SYMBOL(sptlrpc_get_next_secid);
1217
1218 /**************************************************
1219  * client side high-level security APIs           *
1220  **************************************************/
1221
1222 static int sec_cop_flush_ctx_cache(struct ptlrpc_sec *sec, uid_t uid,
1223                                    int grace, int force)
1224 {
1225         struct ptlrpc_sec_policy *policy = sec->ps_policy;
1226
1227         LASSERT(policy->sp_cops);
1228         LASSERT(policy->sp_cops->flush_ctx_cache);
1229
1230         return policy->sp_cops->flush_ctx_cache(sec, uid, grace, force);
1231 }
1232
1233 static void sec_cop_destroy_sec(struct ptlrpc_sec *sec)
1234 {
1235         struct ptlrpc_sec_policy *policy = sec->ps_policy;
1236
1237         LASSERT_ATOMIC_ZERO(&sec->ps_refcount);
1238         LASSERT_ATOMIC_ZERO(&sec->ps_nctx);
1239         LASSERT(policy->sp_cops->destroy_sec);
1240
1241         CDEBUG(D_SEC, "%s@%p: being destroied\n", sec->ps_policy->sp_name, sec);
1242
1243         policy->sp_cops->destroy_sec(sec);
1244         sptlrpc_policy_put(policy);
1245 }
1246
1247 void sptlrpc_sec_destroy(struct ptlrpc_sec *sec)
1248 {
1249         sec_cop_destroy_sec(sec);
1250 }
1251 EXPORT_SYMBOL(sptlrpc_sec_destroy);
1252
1253 static void sptlrpc_sec_kill(struct ptlrpc_sec *sec)
1254 {
1255         LASSERT_ATOMIC_POS(&sec->ps_refcount);
1256
1257         if (sec->ps_policy->sp_cops->kill_sec) {
1258                 sec->ps_policy->sp_cops->kill_sec(sec);
1259
1260                 sec_cop_flush_ctx_cache(sec, -1, 1, 1);
1261         }
1262 }
1263
1264 struct ptlrpc_sec *sptlrpc_sec_get(struct ptlrpc_sec *sec)
1265 {
1266         if (sec)
1267                 cfs_atomic_inc(&sec->ps_refcount);
1268
1269         return sec;
1270 }
1271 EXPORT_SYMBOL(sptlrpc_sec_get);
1272
1273 void sptlrpc_sec_put(struct ptlrpc_sec *sec)
1274 {
1275         if (sec) {
1276                 LASSERT_ATOMIC_POS(&sec->ps_refcount);
1277
1278                 if (cfs_atomic_dec_and_test(&sec->ps_refcount)) {
1279                         sptlrpc_gc_del_sec(sec);
1280                         sec_cop_destroy_sec(sec);
1281                 }
1282         }
1283 }
1284 EXPORT_SYMBOL(sptlrpc_sec_put);
1285
1286 /*
1287  * policy module is responsible for taking refrence of import
1288  */
1289 static
1290 struct ptlrpc_sec * sptlrpc_sec_create(struct obd_import *imp,
1291                                        struct ptlrpc_svc_ctx *svc_ctx,
1292                                        struct sptlrpc_flavor *sf,
1293                                        enum lustre_sec_part sp)
1294 {
1295         struct ptlrpc_sec_policy *policy;
1296         struct ptlrpc_sec        *sec;
1297         char                      str[32];
1298         ENTRY;
1299
1300         if (svc_ctx) {
1301                 LASSERT(imp->imp_dlm_fake == 1);
1302
1303                 CDEBUG(D_SEC, "%s %s: reverse sec using flavor %s\n",
1304                        imp->imp_obd->obd_type->typ_name,
1305                        imp->imp_obd->obd_name,
1306                        sptlrpc_flavor2name(sf, str, sizeof(str)));
1307
1308                 policy = sptlrpc_policy_get(svc_ctx->sc_policy);
1309                 sf->sf_flags |= PTLRPC_SEC_FL_REVERSE | PTLRPC_SEC_FL_ROOTONLY;
1310         } else {
1311                 LASSERT(imp->imp_dlm_fake == 0);
1312
1313                 CDEBUG(D_SEC, "%s %s: select security flavor %s\n",
1314                        imp->imp_obd->obd_type->typ_name,
1315                        imp->imp_obd->obd_name,
1316                        sptlrpc_flavor2name(sf, str, sizeof(str)));
1317
1318                 policy = sptlrpc_wireflavor2policy(sf->sf_rpc);
1319                 if (!policy) {
1320                         CERROR("invalid flavor 0x%x\n", sf->sf_rpc);
1321                         RETURN(NULL);
1322                 }
1323         }
1324
1325         sec = policy->sp_cops->create_sec(imp, svc_ctx, sf);
1326         if (sec) {
1327                 cfs_atomic_inc(&sec->ps_refcount);
1328
1329                 sec->ps_part = sp;
1330
1331                 if (sec->ps_gc_interval && policy->sp_cops->gc_ctx)
1332                         sptlrpc_gc_add_sec(sec);
1333         } else {
1334                 sptlrpc_policy_put(policy);
1335         }
1336
1337         RETURN(sec);
1338 }
1339
1340 struct ptlrpc_sec *sptlrpc_import_sec_ref(struct obd_import *imp)
1341 {
1342         struct ptlrpc_sec *sec;
1343
1344         cfs_spin_lock(&imp->imp_lock);
1345         sec = sptlrpc_sec_get(imp->imp_sec);
1346         cfs_spin_unlock(&imp->imp_lock);
1347
1348         return sec;
1349 }
1350 EXPORT_SYMBOL(sptlrpc_import_sec_ref);
1351
1352 static void sptlrpc_import_sec_install(struct obd_import *imp,
1353                                        struct ptlrpc_sec *sec)
1354 {
1355         struct ptlrpc_sec *old_sec;
1356
1357         LASSERT_ATOMIC_POS(&sec->ps_refcount);
1358
1359         cfs_spin_lock(&imp->imp_lock);
1360         old_sec = imp->imp_sec;
1361         imp->imp_sec = sec;
1362         cfs_spin_unlock(&imp->imp_lock);
1363
1364         if (old_sec) {
1365                 sptlrpc_sec_kill(old_sec);
1366
1367                 /* balance the ref taken by this import */
1368                 sptlrpc_sec_put(old_sec);
1369         }
1370 }
1371
1372 static inline
1373 int flavor_equal(struct sptlrpc_flavor *sf1, struct sptlrpc_flavor *sf2)
1374 {
1375         return (memcmp(sf1, sf2, sizeof(*sf1)) == 0);
1376 }
1377
1378 static inline
1379 void flavor_copy(struct sptlrpc_flavor *dst, struct sptlrpc_flavor *src)
1380 {
1381         *dst = *src;
1382 }
1383
1384 static void sptlrpc_import_sec_adapt_inplace(struct obd_import *imp,
1385                                              struct ptlrpc_sec *sec,
1386                                              struct sptlrpc_flavor *sf)
1387 {
1388         char    str1[32], str2[32];
1389
1390         if (sec->ps_flvr.sf_flags != sf->sf_flags)
1391                 CDEBUG(D_SEC, "changing sec flags: %s -> %s\n",
1392                        sptlrpc_secflags2str(sec->ps_flvr.sf_flags,
1393                                             str1, sizeof(str1)),
1394                        sptlrpc_secflags2str(sf->sf_flags,
1395                                             str2, sizeof(str2)));
1396
1397         cfs_spin_lock(&sec->ps_lock);
1398         flavor_copy(&sec->ps_flvr, sf);
1399         cfs_spin_unlock(&sec->ps_lock);
1400 }
1401
1402 /**
1403  * To get an appropriate ptlrpc_sec for the \a imp, according to the current
1404  * configuration. Upon called, imp->imp_sec may or may not be NULL.
1405  *
1406  *  - regular import: \a svc_ctx should be NULL and \a flvr is ignored;
1407  *  - reverse import: \a svc_ctx and \a flvr are obtained from incoming request.
1408  */
1409 int sptlrpc_import_sec_adapt(struct obd_import *imp,
1410                              struct ptlrpc_svc_ctx *svc_ctx,
1411                              struct sptlrpc_flavor *flvr)
1412 {
1413         struct ptlrpc_connection   *conn;
1414         struct sptlrpc_flavor       sf;
1415         struct ptlrpc_sec          *sec, *newsec;
1416         enum lustre_sec_part        sp;
1417         char                        str[24];
1418         int                         rc = 0;
1419         ENTRY;
1420
1421         cfs_might_sleep();
1422
1423         if (imp == NULL)
1424                 RETURN(0);
1425
1426         conn = imp->imp_connection;
1427
1428         if (svc_ctx == NULL) {
1429                 struct client_obd *cliobd = &imp->imp_obd->u.cli;
1430                 /*
1431                  * normal import, determine flavor from rule set, except
1432                  * for mgc the flavor is predetermined.
1433                  */
1434                 if (cliobd->cl_sp_me == LUSTRE_SP_MGC)
1435                         sf = cliobd->cl_flvr_mgc;
1436                 else
1437                         sptlrpc_conf_choose_flavor(cliobd->cl_sp_me,
1438                                                    cliobd->cl_sp_to,
1439                                                    &cliobd->cl_target_uuid,
1440                                                    conn->c_self, &sf);
1441
1442                 sp = imp->imp_obd->u.cli.cl_sp_me;
1443         } else {
1444                 /* reverse import, determine flavor from incoming reqeust */
1445                 sf = *flvr;
1446
1447                 if (sf.sf_rpc != SPTLRPC_FLVR_NULL)
1448                         sf.sf_flags = PTLRPC_SEC_FL_REVERSE |
1449                                       PTLRPC_SEC_FL_ROOTONLY;
1450
1451                 sp = sptlrpc_target_sec_part(imp->imp_obd);
1452         }
1453
1454         sec = sptlrpc_import_sec_ref(imp);
1455         if (sec) {
1456                 char    str2[24];
1457
1458                 if (flavor_equal(&sf, &sec->ps_flvr))
1459                         GOTO(out, rc);
1460
1461                 CDEBUG(D_SEC, "import %s->%s: changing flavor %s -> %s\n",
1462                        imp->imp_obd->obd_name,
1463                        obd_uuid2str(&conn->c_remote_uuid),
1464                        sptlrpc_flavor2name(&sec->ps_flvr, str, sizeof(str)),
1465                        sptlrpc_flavor2name(&sf, str2, sizeof(str2)));
1466
1467                 if (SPTLRPC_FLVR_POLICY(sf.sf_rpc) ==
1468                     SPTLRPC_FLVR_POLICY(sec->ps_flvr.sf_rpc) &&
1469                     SPTLRPC_FLVR_MECH(sf.sf_rpc) ==
1470                     SPTLRPC_FLVR_MECH(sec->ps_flvr.sf_rpc)) {
1471                         sptlrpc_import_sec_adapt_inplace(imp, sec, &sf);
1472                         GOTO(out, rc);
1473                 }
1474         } else if (SPTLRPC_FLVR_BASE(sf.sf_rpc) !=
1475                    SPTLRPC_FLVR_BASE(SPTLRPC_FLVR_NULL)) {
1476                 CDEBUG(D_SEC, "import %s->%s netid %x: select flavor %s\n",
1477                        imp->imp_obd->obd_name,
1478                        obd_uuid2str(&conn->c_remote_uuid),
1479                        LNET_NIDNET(conn->c_self),
1480                        sptlrpc_flavor2name(&sf, str, sizeof(str)));
1481         }
1482
1483         cfs_mutex_lock(&imp->imp_sec_mutex);
1484
1485         newsec = sptlrpc_sec_create(imp, svc_ctx, &sf, sp);
1486         if (newsec) {
1487                 sptlrpc_import_sec_install(imp, newsec);
1488         } else {
1489                 CERROR("import %s->%s: failed to create new sec\n",
1490                        imp->imp_obd->obd_name,
1491                        obd_uuid2str(&conn->c_remote_uuid));
1492                 rc = -EPERM;
1493         }
1494
1495         cfs_mutex_unlock(&imp->imp_sec_mutex);
1496 out:
1497         sptlrpc_sec_put(sec);
1498         RETURN(rc);
1499 }
1500
1501 void sptlrpc_import_sec_put(struct obd_import *imp)
1502 {
1503         if (imp->imp_sec) {
1504                 sptlrpc_sec_kill(imp->imp_sec);
1505
1506                 sptlrpc_sec_put(imp->imp_sec);
1507                 imp->imp_sec = NULL;
1508         }
1509 }
1510
1511 static void import_flush_ctx_common(struct obd_import *imp,
1512                                     uid_t uid, int grace, int force)
1513 {
1514         struct ptlrpc_sec *sec;
1515
1516         if (imp == NULL)
1517                 return;
1518
1519         sec = sptlrpc_import_sec_ref(imp);
1520         if (sec == NULL)
1521                 return;
1522
1523         sec_cop_flush_ctx_cache(sec, uid, grace, force);
1524         sptlrpc_sec_put(sec);
1525 }
1526
1527 void sptlrpc_import_flush_root_ctx(struct obd_import *imp)
1528 {
1529         /* it's important to use grace mode, see explain in
1530          * sptlrpc_req_refresh_ctx() */
1531         import_flush_ctx_common(imp, 0, 1, 1);
1532 }
1533
1534 void sptlrpc_import_flush_my_ctx(struct obd_import *imp)
1535 {
1536         import_flush_ctx_common(imp, cfs_curproc_uid(), 1, 1);
1537 }
1538 EXPORT_SYMBOL(sptlrpc_import_flush_my_ctx);
1539
1540 void sptlrpc_import_flush_all_ctx(struct obd_import *imp)
1541 {
1542         import_flush_ctx_common(imp, -1, 1, 1);
1543 }
1544 EXPORT_SYMBOL(sptlrpc_import_flush_all_ctx);
1545
1546 /**
1547  * Used by ptlrpc client to allocate request buffer of \a req. Upon return
1548  * successfully, req->rq_reqmsg points to a buffer with size \a msgsize.
1549  */
1550 int sptlrpc_cli_alloc_reqbuf(struct ptlrpc_request *req, int msgsize)
1551 {
1552         struct ptlrpc_cli_ctx *ctx = req->rq_cli_ctx;
1553         struct ptlrpc_sec_policy *policy;
1554         int rc;
1555
1556         LASSERT(ctx);
1557         LASSERT(ctx->cc_sec);
1558         LASSERT(ctx->cc_sec->ps_policy);
1559         LASSERT(req->rq_reqmsg == NULL);
1560         LASSERT_ATOMIC_POS(&ctx->cc_refcount);
1561
1562         policy = ctx->cc_sec->ps_policy;
1563         rc = policy->sp_cops->alloc_reqbuf(ctx->cc_sec, req, msgsize);
1564         if (!rc) {
1565                 LASSERT(req->rq_reqmsg);
1566                 LASSERT(req->rq_reqbuf || req->rq_clrbuf);
1567
1568                 /* zeroing preallocated buffer */
1569                 if (req->rq_pool)
1570                         memset(req->rq_reqmsg, 0, msgsize);
1571         }
1572
1573         return rc;
1574 }
1575
1576 /**
1577  * Used by ptlrpc client to free request buffer of \a req. After this
1578  * req->rq_reqmsg is set to NULL and should not be accessed anymore.
1579  */
1580 void sptlrpc_cli_free_reqbuf(struct ptlrpc_request *req)
1581 {
1582         struct ptlrpc_cli_ctx *ctx = req->rq_cli_ctx;
1583         struct ptlrpc_sec_policy *policy;
1584
1585         LASSERT(ctx);
1586         LASSERT(ctx->cc_sec);
1587         LASSERT(ctx->cc_sec->ps_policy);
1588         LASSERT_ATOMIC_POS(&ctx->cc_refcount);
1589
1590         if (req->rq_reqbuf == NULL && req->rq_clrbuf == NULL)
1591                 return;
1592
1593         policy = ctx->cc_sec->ps_policy;
1594         policy->sp_cops->free_reqbuf(ctx->cc_sec, req);
1595         req->rq_reqmsg = NULL;
1596 }
1597
1598 /*
1599  * NOTE caller must guarantee the buffer size is enough for the enlargement
1600  */
1601 void _sptlrpc_enlarge_msg_inplace(struct lustre_msg *msg,
1602                                   int segment, int newsize)
1603 {
1604         void   *src, *dst;
1605         int     oldsize, oldmsg_size, movesize;
1606
1607         LASSERT(segment < msg->lm_bufcount);
1608         LASSERT(msg->lm_buflens[segment] <= newsize);
1609
1610         if (msg->lm_buflens[segment] == newsize)
1611                 return;
1612
1613         /* nothing to do if we are enlarging the last segment */
1614         if (segment == msg->lm_bufcount - 1) {
1615                 msg->lm_buflens[segment] = newsize;
1616                 return;
1617         }
1618
1619         oldsize = msg->lm_buflens[segment];
1620
1621         src = lustre_msg_buf(msg, segment + 1, 0);
1622         msg->lm_buflens[segment] = newsize;
1623         dst = lustre_msg_buf(msg, segment + 1, 0);
1624         msg->lm_buflens[segment] = oldsize;
1625
1626         /* move from segment + 1 to end segment */
1627         LASSERT(msg->lm_magic == LUSTRE_MSG_MAGIC_V2);
1628         oldmsg_size = lustre_msg_size_v2(msg->lm_bufcount, msg->lm_buflens);
1629         movesize = oldmsg_size - ((unsigned long) src - (unsigned long) msg);
1630         LASSERT(movesize >= 0);
1631
1632         if (movesize)
1633                 memmove(dst, src, movesize);
1634
1635         /* note we don't clear the ares where old data live, not secret */
1636
1637         /* finally set new segment size */
1638         msg->lm_buflens[segment] = newsize;
1639 }
1640 EXPORT_SYMBOL(_sptlrpc_enlarge_msg_inplace);
1641
1642 /**
1643  * Used by ptlrpc client to enlarge the \a segment of request message pointed
1644  * by req->rq_reqmsg to size \a newsize, all previously filled-in data will be
1645  * preserved after the enlargement. this must be called after original request
1646  * buffer being allocated.
1647  *
1648  * \note after this be called, rq_reqmsg and rq_reqlen might have been changed,
1649  * so caller should refresh its local pointers if needed.
1650  */
1651 int sptlrpc_cli_enlarge_reqbuf(struct ptlrpc_request *req,
1652                                int segment, int newsize)
1653 {
1654         struct ptlrpc_cli_ctx    *ctx = req->rq_cli_ctx;
1655         struct ptlrpc_sec_cops   *cops;
1656         struct lustre_msg        *msg = req->rq_reqmsg;
1657
1658         LASSERT(ctx);
1659         LASSERT(msg);
1660         LASSERT(msg->lm_bufcount > segment);
1661         LASSERT(msg->lm_buflens[segment] <= newsize);
1662
1663         if (msg->lm_buflens[segment] == newsize)
1664                 return 0;
1665
1666         cops = ctx->cc_sec->ps_policy->sp_cops;
1667         LASSERT(cops->enlarge_reqbuf);
1668         return cops->enlarge_reqbuf(ctx->cc_sec, req, segment, newsize);
1669 }
1670 EXPORT_SYMBOL(sptlrpc_cli_enlarge_reqbuf);
1671
1672 /**
1673  * Used by ptlrpc client to allocate reply buffer of \a req.
1674  *
1675  * \note After this, req->rq_repmsg is still not accessible.
1676  */
1677 int sptlrpc_cli_alloc_repbuf(struct ptlrpc_request *req, int msgsize)
1678 {
1679         struct ptlrpc_cli_ctx *ctx = req->rq_cli_ctx;
1680         struct ptlrpc_sec_policy *policy;
1681         ENTRY;
1682
1683         LASSERT(ctx);
1684         LASSERT(ctx->cc_sec);
1685         LASSERT(ctx->cc_sec->ps_policy);
1686
1687         if (req->rq_repbuf)
1688                 RETURN(0);
1689
1690         policy = ctx->cc_sec->ps_policy;
1691         RETURN(policy->sp_cops->alloc_repbuf(ctx->cc_sec, req, msgsize));
1692 }
1693
1694 /**
1695  * Used by ptlrpc client to free reply buffer of \a req. After this
1696  * req->rq_repmsg is set to NULL and should not be accessed anymore.
1697  */
1698 void sptlrpc_cli_free_repbuf(struct ptlrpc_request *req)
1699 {
1700         struct ptlrpc_cli_ctx *ctx = req->rq_cli_ctx;
1701         struct ptlrpc_sec_policy *policy;
1702         ENTRY;
1703
1704         LASSERT(ctx);
1705         LASSERT(ctx->cc_sec);
1706         LASSERT(ctx->cc_sec->ps_policy);
1707         LASSERT_ATOMIC_POS(&ctx->cc_refcount);
1708
1709         if (req->rq_repbuf == NULL)
1710                 return;
1711         LASSERT(req->rq_repbuf_len);
1712
1713         policy = ctx->cc_sec->ps_policy;
1714         policy->sp_cops->free_repbuf(ctx->cc_sec, req);
1715         req->rq_repmsg = NULL;
1716         EXIT;
1717 }
1718
1719 int sptlrpc_cli_install_rvs_ctx(struct obd_import *imp,
1720                                 struct ptlrpc_cli_ctx *ctx)
1721 {
1722         struct ptlrpc_sec_policy *policy = ctx->cc_sec->ps_policy;
1723
1724         if (!policy->sp_cops->install_rctx)
1725                 return 0;
1726         return policy->sp_cops->install_rctx(imp, ctx->cc_sec, ctx);
1727 }
1728
1729 int sptlrpc_svc_install_rvs_ctx(struct obd_import *imp,
1730                                 struct ptlrpc_svc_ctx *ctx)
1731 {
1732         struct ptlrpc_sec_policy *policy = ctx->sc_policy;
1733
1734         if (!policy->sp_sops->install_rctx)
1735                 return 0;
1736         return policy->sp_sops->install_rctx(imp, ctx);
1737 }
1738
1739 /****************************************
1740  * server side security                 *
1741  ****************************************/
1742
1743 static int flavor_allowed(struct sptlrpc_flavor *exp,
1744                           struct ptlrpc_request *req)
1745 {
1746         struct sptlrpc_flavor *flvr = &req->rq_flvr;
1747
1748         if (exp->sf_rpc == SPTLRPC_FLVR_ANY || exp->sf_rpc == flvr->sf_rpc)
1749                 return 1;
1750
1751         if ((req->rq_ctx_init || req->rq_ctx_fini) &&
1752             SPTLRPC_FLVR_POLICY(exp->sf_rpc) ==
1753             SPTLRPC_FLVR_POLICY(flvr->sf_rpc) &&
1754             SPTLRPC_FLVR_MECH(exp->sf_rpc) == SPTLRPC_FLVR_MECH(flvr->sf_rpc))
1755                 return 1;
1756
1757         return 0;
1758 }
1759
1760 #define EXP_FLVR_UPDATE_EXPIRE      (OBD_TIMEOUT_DEFAULT + 10)
1761
1762 /**
1763  * Given an export \a exp, check whether the flavor of incoming \a req
1764  * is allowed by the export \a exp. Main logic is about taking care of
1765  * changing configurations. Return 0 means success.
1766  */
1767 int sptlrpc_target_export_check(struct obd_export *exp,
1768                                 struct ptlrpc_request *req)
1769 {
1770         struct sptlrpc_flavor   flavor;
1771
1772         if (exp == NULL)
1773                 return 0;
1774
1775         /* client side export has no imp_reverse, skip
1776          * FIXME maybe we should check flavor this as well??? */
1777         if (exp->exp_imp_reverse == NULL)
1778                 return 0;
1779
1780         /* don't care about ctx fini rpc */
1781         if (req->rq_ctx_fini)
1782                 return 0;
1783
1784         cfs_spin_lock(&exp->exp_lock);
1785
1786         /* if flavor just changed (exp->exp_flvr_changed != 0), we wait for
1787          * the first req with the new flavor, then treat it as current flavor,
1788          * adapt reverse sec according to it.
1789          * note the first rpc with new flavor might not be with root ctx, in
1790          * which case delay the sec_adapt by leaving exp_flvr_adapt == 1. */
1791         if (unlikely(exp->exp_flvr_changed) &&
1792             flavor_allowed(&exp->exp_flvr_old[1], req)) {
1793                 /* make the new flavor as "current", and old ones as
1794                  * about-to-expire */
1795                 CDEBUG(D_SEC, "exp %p: just changed: %x->%x\n", exp,
1796                        exp->exp_flvr.sf_rpc, exp->exp_flvr_old[1].sf_rpc);
1797                 flavor = exp->exp_flvr_old[1];
1798                 exp->exp_flvr_old[1] = exp->exp_flvr_old[0];
1799                 exp->exp_flvr_expire[1] = exp->exp_flvr_expire[0];
1800                 exp->exp_flvr_old[0] = exp->exp_flvr;
1801                 exp->exp_flvr_expire[0] = cfs_time_current_sec() +
1802                                           EXP_FLVR_UPDATE_EXPIRE;
1803                 exp->exp_flvr = flavor;
1804
1805                 /* flavor change finished */
1806                 exp->exp_flvr_changed = 0;
1807                 LASSERT(exp->exp_flvr_adapt == 1);
1808
1809                 /* if it's gss, we only interested in root ctx init */
1810                 if (req->rq_auth_gss &&
1811                     !(req->rq_ctx_init &&
1812                       (req->rq_auth_usr_root || req->rq_auth_usr_mdt ||
1813                        req->rq_auth_usr_ost))) {
1814                         cfs_spin_unlock(&exp->exp_lock);
1815                         CDEBUG(D_SEC, "is good but not root(%d:%d:%d:%d:%d)\n",
1816                                req->rq_auth_gss, req->rq_ctx_init,
1817                                req->rq_auth_usr_root, req->rq_auth_usr_mdt,
1818                                req->rq_auth_usr_ost);
1819                         return 0;
1820                 }
1821
1822                 exp->exp_flvr_adapt = 0;
1823                 cfs_spin_unlock(&exp->exp_lock);
1824
1825                 return sptlrpc_import_sec_adapt(exp->exp_imp_reverse,
1826                                                 req->rq_svc_ctx, &flavor);
1827         }
1828
1829         /* if it equals to the current flavor, we accept it, but need to
1830          * dealing with reverse sec/ctx */
1831         if (likely(flavor_allowed(&exp->exp_flvr, req))) {
1832                 /* most cases should return here, we only interested in
1833                  * gss root ctx init */
1834                 if (!req->rq_auth_gss || !req->rq_ctx_init ||
1835                     (!req->rq_auth_usr_root && !req->rq_auth_usr_mdt &&
1836                      !req->rq_auth_usr_ost)) {
1837                         cfs_spin_unlock(&exp->exp_lock);
1838                         return 0;
1839                 }
1840
1841                 /* if flavor just changed, we should not proceed, just leave
1842                  * it and current flavor will be discovered and replaced
1843                  * shortly, and let _this_ rpc pass through */
1844                 if (exp->exp_flvr_changed) {
1845                         LASSERT(exp->exp_flvr_adapt);
1846                         cfs_spin_unlock(&exp->exp_lock);
1847                         return 0;
1848                 }
1849
1850                 if (exp->exp_flvr_adapt) {
1851                         exp->exp_flvr_adapt = 0;
1852                         CDEBUG(D_SEC, "exp %p (%x|%x|%x): do delayed adapt\n",
1853                                exp, exp->exp_flvr.sf_rpc,
1854                                exp->exp_flvr_old[0].sf_rpc,
1855                                exp->exp_flvr_old[1].sf_rpc);
1856                         flavor = exp->exp_flvr;
1857                         cfs_spin_unlock(&exp->exp_lock);
1858
1859                         return sptlrpc_import_sec_adapt(exp->exp_imp_reverse,
1860                                                         req->rq_svc_ctx,
1861                                                         &flavor);
1862                 } else {
1863                         CDEBUG(D_SEC, "exp %p (%x|%x|%x): is current flavor, "
1864                                "install rvs ctx\n", exp, exp->exp_flvr.sf_rpc,
1865                                exp->exp_flvr_old[0].sf_rpc,
1866                                exp->exp_flvr_old[1].sf_rpc);
1867                         cfs_spin_unlock(&exp->exp_lock);
1868
1869                         return sptlrpc_svc_install_rvs_ctx(exp->exp_imp_reverse,
1870                                                            req->rq_svc_ctx);
1871                 }
1872         }
1873
1874         if (exp->exp_flvr_expire[0]) {
1875                 if (exp->exp_flvr_expire[0] >= cfs_time_current_sec()) {
1876                         if (flavor_allowed(&exp->exp_flvr_old[0], req)) {
1877                                 CDEBUG(D_SEC, "exp %p (%x|%x|%x): match the "
1878                                        "middle one ("CFS_DURATION_T")\n", exp,
1879                                        exp->exp_flvr.sf_rpc,
1880                                        exp->exp_flvr_old[0].sf_rpc,
1881                                        exp->exp_flvr_old[1].sf_rpc,
1882                                        exp->exp_flvr_expire[0] -
1883                                                 cfs_time_current_sec());
1884                                 cfs_spin_unlock(&exp->exp_lock);
1885                                 return 0;
1886                         }
1887                 } else {
1888                         CDEBUG(D_SEC, "mark middle expired\n");
1889                         exp->exp_flvr_expire[0] = 0;
1890                 }
1891                 CDEBUG(D_SEC, "exp %p (%x|%x|%x): %x not match middle\n", exp,
1892                        exp->exp_flvr.sf_rpc,
1893                        exp->exp_flvr_old[0].sf_rpc, exp->exp_flvr_old[1].sf_rpc,
1894                        req->rq_flvr.sf_rpc);
1895         }
1896
1897         /* now it doesn't match the current flavor, the only chance we can
1898          * accept it is match the old flavors which is not expired. */
1899         if (exp->exp_flvr_changed == 0 && exp->exp_flvr_expire[1]) {
1900                 if (exp->exp_flvr_expire[1] >= cfs_time_current_sec()) {
1901                         if (flavor_allowed(&exp->exp_flvr_old[1], req)) {
1902                                 CDEBUG(D_SEC, "exp %p (%x|%x|%x): match the "
1903                                        "oldest one ("CFS_DURATION_T")\n", exp,
1904                                        exp->exp_flvr.sf_rpc,
1905                                        exp->exp_flvr_old[0].sf_rpc,
1906                                        exp->exp_flvr_old[1].sf_rpc,
1907                                        exp->exp_flvr_expire[1] -
1908                                                 cfs_time_current_sec());
1909                                 cfs_spin_unlock(&exp->exp_lock);
1910                                 return 0;
1911                         }
1912                 } else {
1913                         CDEBUG(D_SEC, "mark oldest expired\n");
1914                         exp->exp_flvr_expire[1] = 0;
1915                 }
1916                 CDEBUG(D_SEC, "exp %p (%x|%x|%x): %x not match found\n",
1917                        exp, exp->exp_flvr.sf_rpc,
1918                        exp->exp_flvr_old[0].sf_rpc, exp->exp_flvr_old[1].sf_rpc,
1919                        req->rq_flvr.sf_rpc);
1920         } else {
1921                 CDEBUG(D_SEC, "exp %p (%x|%x|%x): skip the last one\n",
1922                        exp, exp->exp_flvr.sf_rpc, exp->exp_flvr_old[0].sf_rpc,
1923                        exp->exp_flvr_old[1].sf_rpc);
1924         }
1925
1926         cfs_spin_unlock(&exp->exp_lock);
1927
1928         CWARN("exp %p(%s): req %p (%u|%u|%u|%u|%u|%u) with "
1929               "unauthorized flavor %x, expect %x|%x(%+ld)|%x(%+ld)\n",
1930               exp, exp->exp_obd->obd_name,
1931               req, req->rq_auth_gss, req->rq_ctx_init, req->rq_ctx_fini,
1932               req->rq_auth_usr_root, req->rq_auth_usr_mdt, req->rq_auth_usr_ost,
1933               req->rq_flvr.sf_rpc,
1934               exp->exp_flvr.sf_rpc,
1935               exp->exp_flvr_old[0].sf_rpc,
1936               exp->exp_flvr_expire[0] ?
1937               (unsigned long) (exp->exp_flvr_expire[0] -
1938                                cfs_time_current_sec()) : 0,
1939               exp->exp_flvr_old[1].sf_rpc,
1940               exp->exp_flvr_expire[1] ?
1941               (unsigned long) (exp->exp_flvr_expire[1] -
1942                                cfs_time_current_sec()) : 0);
1943         return -EACCES;
1944 }
1945 EXPORT_SYMBOL(sptlrpc_target_export_check);
1946
1947 void sptlrpc_target_update_exp_flavor(struct obd_device *obd,
1948                                       struct sptlrpc_rule_set *rset)
1949 {
1950         struct obd_export       *exp;
1951         struct sptlrpc_flavor    new_flvr;
1952
1953         LASSERT(obd);
1954
1955         cfs_spin_lock(&obd->obd_dev_lock);
1956
1957         cfs_list_for_each_entry(exp, &obd->obd_exports, exp_obd_chain) {
1958                 if (exp->exp_connection == NULL)
1959                         continue;
1960
1961                 /* note if this export had just been updated flavor
1962                  * (exp_flvr_changed == 1), this will override the
1963                  * previous one. */
1964                 cfs_spin_lock(&exp->exp_lock);
1965                 sptlrpc_target_choose_flavor(rset, exp->exp_sp_peer,
1966                                              exp->exp_connection->c_peer.nid,
1967                                              &new_flvr);
1968                 if (exp->exp_flvr_changed ||
1969                     !flavor_equal(&new_flvr, &exp->exp_flvr)) {
1970                         exp->exp_flvr_old[1] = new_flvr;
1971                         exp->exp_flvr_expire[1] = 0;
1972                         exp->exp_flvr_changed = 1;
1973                         exp->exp_flvr_adapt = 1;
1974
1975                         CDEBUG(D_SEC, "exp %p (%s): updated flavor %x->%x\n",
1976                                exp, sptlrpc_part2name(exp->exp_sp_peer),
1977                                exp->exp_flvr.sf_rpc,
1978                                exp->exp_flvr_old[1].sf_rpc);
1979                 }
1980                 cfs_spin_unlock(&exp->exp_lock);
1981         }
1982
1983         cfs_spin_unlock(&obd->obd_dev_lock);
1984 }
1985 EXPORT_SYMBOL(sptlrpc_target_update_exp_flavor);
1986
1987 static int sptlrpc_svc_check_from(struct ptlrpc_request *req, int svc_rc)
1988 {
1989         /* peer's claim is unreliable unless gss is being used */
1990         if (!req->rq_auth_gss || svc_rc == SECSVC_DROP)
1991                 return svc_rc;
1992
1993         switch (req->rq_sp_from) {
1994         case LUSTRE_SP_CLI:
1995                 if (req->rq_auth_usr_mdt || req->rq_auth_usr_ost) {
1996                         DEBUG_REQ(D_ERROR, req, "faked source CLI");
1997                         svc_rc = SECSVC_DROP;
1998                 }
1999                 break;
2000         case LUSTRE_SP_MDT:
2001                 if (!req->rq_auth_usr_mdt) {
2002                         DEBUG_REQ(D_ERROR, req, "faked source MDT");
2003                         svc_rc = SECSVC_DROP;
2004                 }
2005                 break;
2006         case LUSTRE_SP_OST:
2007                 if (!req->rq_auth_usr_ost) {
2008                         DEBUG_REQ(D_ERROR, req, "faked source OST");
2009                         svc_rc = SECSVC_DROP;
2010                 }
2011                 break;
2012         case LUSTRE_SP_MGS:
2013         case LUSTRE_SP_MGC:
2014                 if (!req->rq_auth_usr_root && !req->rq_auth_usr_mdt &&
2015                     !req->rq_auth_usr_ost) {
2016                         DEBUG_REQ(D_ERROR, req, "faked source MGC/MGS");
2017                         svc_rc = SECSVC_DROP;
2018                 }
2019                 break;
2020         case LUSTRE_SP_ANY:
2021         default:
2022                 DEBUG_REQ(D_ERROR, req, "invalid source %u", req->rq_sp_from);
2023                 svc_rc = SECSVC_DROP;
2024         }
2025
2026         return svc_rc;
2027 }
2028
2029 /**
2030  * Used by ptlrpc server, to perform transformation upon request message of
2031  * incoming \a req. This must be the first thing to do with a incoming
2032  * request in ptlrpc layer.
2033  *
2034  * \retval SECSVC_OK success, and req->rq_reqmsg point to request message in
2035  * clear text, size is req->rq_reqlen; also req->rq_svc_ctx is set.
2036  * \retval SECSVC_COMPLETE success, the request has been fully processed, and
2037  * reply message has been prepared.
2038  * \retval SECSVC_DROP failed, this request should be dropped.
2039  */
2040 int sptlrpc_svc_unwrap_request(struct ptlrpc_request *req)
2041 {
2042         struct ptlrpc_sec_policy *policy;
2043         struct lustre_msg        *msg = req->rq_reqbuf;
2044         int                       rc;
2045         ENTRY;
2046
2047         LASSERT(msg);
2048         LASSERT(req->rq_reqmsg == NULL);
2049         LASSERT(req->rq_repmsg == NULL);
2050         LASSERT(req->rq_svc_ctx == NULL);
2051
2052         req->rq_req_swab_mask = 0;
2053
2054         rc = __lustre_unpack_msg(msg, req->rq_reqdata_len);
2055         switch (rc) {
2056         case 1:
2057                 lustre_set_req_swabbed(req, MSG_PTLRPC_HEADER_OFF);
2058         case 0:
2059                 break;
2060         default:
2061                 CERROR("error unpacking request from %s x"LPU64"\n",
2062                        libcfs_id2str(req->rq_peer), req->rq_xid);
2063                 RETURN(SECSVC_DROP);
2064         }
2065
2066         req->rq_flvr.sf_rpc = WIRE_FLVR(msg->lm_secflvr);
2067         req->rq_sp_from = LUSTRE_SP_ANY;
2068         req->rq_auth_uid = INVALID_UID;
2069         req->rq_auth_mapped_uid = INVALID_UID;
2070
2071         policy = sptlrpc_wireflavor2policy(req->rq_flvr.sf_rpc);
2072         if (!policy) {
2073                 CERROR("unsupported rpc flavor %x\n", req->rq_flvr.sf_rpc);
2074                 RETURN(SECSVC_DROP);
2075         }
2076
2077         LASSERT(policy->sp_sops->accept);
2078         rc = policy->sp_sops->accept(req);
2079         sptlrpc_policy_put(policy);
2080         LASSERT(req->rq_reqmsg || rc != SECSVC_OK);
2081         LASSERT(req->rq_svc_ctx || rc == SECSVC_DROP);
2082
2083         /*
2084          * if it's not null flavor (which means embedded packing msg),
2085          * reset the swab mask for the comming inner msg unpacking.
2086          */
2087         if (SPTLRPC_FLVR_POLICY(req->rq_flvr.sf_rpc) != SPTLRPC_POLICY_NULL)
2088                 req->rq_req_swab_mask = 0;
2089
2090         /* sanity check for the request source */
2091         rc = sptlrpc_svc_check_from(req, rc);
2092         RETURN(rc);
2093 }
2094
2095 /**
2096  * Used by ptlrpc server, to allocate reply buffer for \a req. If succeed,
2097  * req->rq_reply_state is set, and req->rq_reply_state->rs_msg point to
2098  * a buffer of \a msglen size.
2099  */
2100 int sptlrpc_svc_alloc_rs(struct ptlrpc_request *req, int msglen)
2101 {
2102         struct ptlrpc_sec_policy *policy;
2103         struct ptlrpc_reply_state *rs;
2104         int rc;
2105         ENTRY;
2106
2107         LASSERT(req->rq_svc_ctx);
2108         LASSERT(req->rq_svc_ctx->sc_policy);
2109
2110         policy = req->rq_svc_ctx->sc_policy;
2111         LASSERT(policy->sp_sops->alloc_rs);
2112
2113         rc = policy->sp_sops->alloc_rs(req, msglen);
2114         if (unlikely(rc == -ENOMEM)) {
2115                 /* failed alloc, try emergency pool */
2116                 rs = lustre_get_emerg_rs(req->rq_rqbd->rqbd_service);
2117                 if (rs == NULL)
2118                         RETURN(-ENOMEM);
2119
2120                 req->rq_reply_state = rs;
2121                 rc = policy->sp_sops->alloc_rs(req, msglen);
2122                 if (rc) {
2123                         lustre_put_emerg_rs(rs);
2124                         req->rq_reply_state = NULL;
2125                 }
2126         }
2127
2128         LASSERT(rc != 0 ||
2129                 (req->rq_reply_state && req->rq_reply_state->rs_msg));
2130
2131         RETURN(rc);
2132 }
2133
2134 /**
2135  * Used by ptlrpc server, to perform transformation upon reply message.
2136  *
2137  * \post req->rq_reply_off is set to approriate server-controlled reply offset.
2138  * \post req->rq_repmsg and req->rq_reply_state->rs_msg becomes inaccessible.
2139  */
2140 int sptlrpc_svc_wrap_reply(struct ptlrpc_request *req)
2141 {
2142         struct ptlrpc_sec_policy *policy;
2143         int rc;
2144         ENTRY;
2145
2146         LASSERT(req->rq_svc_ctx);
2147         LASSERT(req->rq_svc_ctx->sc_policy);
2148
2149         policy = req->rq_svc_ctx->sc_policy;
2150         LASSERT(policy->sp_sops->authorize);
2151
2152         rc = policy->sp_sops->authorize(req);
2153         LASSERT(rc || req->rq_reply_state->rs_repdata_len);
2154
2155         RETURN(rc);
2156 }
2157
2158 /**
2159  * Used by ptlrpc server, to free reply_state.
2160  */
2161 void sptlrpc_svc_free_rs(struct ptlrpc_reply_state *rs)
2162 {
2163         struct ptlrpc_sec_policy *policy;
2164         unsigned int prealloc;
2165         ENTRY;
2166
2167         LASSERT(rs->rs_svc_ctx);
2168         LASSERT(rs->rs_svc_ctx->sc_policy);
2169
2170         policy = rs->rs_svc_ctx->sc_policy;
2171         LASSERT(policy->sp_sops->free_rs);
2172
2173         prealloc = rs->rs_prealloc;
2174         policy->sp_sops->free_rs(rs);
2175
2176         if (prealloc)
2177                 lustre_put_emerg_rs(rs);
2178         EXIT;
2179 }
2180
2181 void sptlrpc_svc_ctx_addref(struct ptlrpc_request *req)
2182 {
2183         struct ptlrpc_svc_ctx *ctx = req->rq_svc_ctx;
2184
2185         if (ctx != NULL)
2186                 cfs_atomic_inc(&ctx->sc_refcount);
2187 }
2188
2189 void sptlrpc_svc_ctx_decref(struct ptlrpc_request *req)
2190 {
2191         struct ptlrpc_svc_ctx *ctx = req->rq_svc_ctx;
2192
2193         if (ctx == NULL)
2194                 return;
2195
2196         LASSERT_ATOMIC_POS(&ctx->sc_refcount);
2197         if (cfs_atomic_dec_and_test(&ctx->sc_refcount)) {
2198                 if (ctx->sc_policy->sp_sops->free_ctx)
2199                         ctx->sc_policy->sp_sops->free_ctx(ctx);
2200         }
2201         req->rq_svc_ctx = NULL;
2202 }
2203
2204 void sptlrpc_svc_ctx_invalidate(struct ptlrpc_request *req)
2205 {
2206         struct ptlrpc_svc_ctx *ctx = req->rq_svc_ctx;
2207
2208         if (ctx == NULL)
2209                 return;
2210
2211         LASSERT_ATOMIC_POS(&ctx->sc_refcount);
2212         if (ctx->sc_policy->sp_sops->invalidate_ctx)
2213                 ctx->sc_policy->sp_sops->invalidate_ctx(ctx);
2214 }
2215 EXPORT_SYMBOL(sptlrpc_svc_ctx_invalidate);
2216
2217 /****************************************
2218  * bulk security                        *
2219  ****************************************/
2220
2221 /**
2222  * Perform transformation upon bulk data pointed by \a desc. This is called
2223  * before transforming the request message.
2224  */
2225 int sptlrpc_cli_wrap_bulk(struct ptlrpc_request *req,
2226                           struct ptlrpc_bulk_desc *desc)
2227 {
2228         struct ptlrpc_cli_ctx *ctx;
2229
2230         LASSERT(req->rq_bulk_read || req->rq_bulk_write);
2231
2232         if (!req->rq_pack_bulk)
2233                 return 0;
2234
2235         ctx = req->rq_cli_ctx;
2236         if (ctx->cc_ops->wrap_bulk)
2237                 return ctx->cc_ops->wrap_bulk(ctx, req, desc);
2238         return 0;
2239 }
2240 EXPORT_SYMBOL(sptlrpc_cli_wrap_bulk);
2241
2242 /**
2243  * This is called after unwrap the reply message.
2244  * return nob of actual plain text size received, or error code.
2245  */
2246 int sptlrpc_cli_unwrap_bulk_read(struct ptlrpc_request *req,
2247                                  struct ptlrpc_bulk_desc *desc,
2248                                  int nob)
2249 {
2250         struct ptlrpc_cli_ctx  *ctx;
2251         int                     rc;
2252
2253         LASSERT(req->rq_bulk_read && !req->rq_bulk_write);
2254
2255         if (!req->rq_pack_bulk)
2256                 return desc->bd_nob_transferred;
2257
2258         ctx = req->rq_cli_ctx;
2259         if (ctx->cc_ops->unwrap_bulk) {
2260                 rc = ctx->cc_ops->unwrap_bulk(ctx, req, desc);
2261                 if (rc < 0)
2262                         return rc;
2263         }
2264         return desc->bd_nob_transferred;
2265 }
2266 EXPORT_SYMBOL(sptlrpc_cli_unwrap_bulk_read);
2267
2268 /**
2269  * This is called after unwrap the reply message.
2270  * return 0 for success or error code.
2271  */
2272 int sptlrpc_cli_unwrap_bulk_write(struct ptlrpc_request *req,
2273                                   struct ptlrpc_bulk_desc *desc)
2274 {
2275         struct ptlrpc_cli_ctx  *ctx;
2276         int                     rc;
2277
2278         LASSERT(!req->rq_bulk_read && req->rq_bulk_write);
2279
2280         if (!req->rq_pack_bulk)
2281                 return 0;
2282
2283         ctx = req->rq_cli_ctx;
2284         if (ctx->cc_ops->unwrap_bulk) {
2285                 rc = ctx->cc_ops->unwrap_bulk(ctx, req, desc);
2286                 if (rc < 0)
2287                         return rc;
2288         }
2289
2290         /*
2291          * if everything is going right, nob should equals to nob_transferred.
2292          * in case of privacy mode, nob_transferred needs to be adjusted.
2293          */
2294         if (desc->bd_nob != desc->bd_nob_transferred) {
2295                 CERROR("nob %d doesn't match transferred nob %d",
2296                        desc->bd_nob, desc->bd_nob_transferred);
2297                 return -EPROTO;
2298         }
2299
2300         return 0;
2301 }
2302 EXPORT_SYMBOL(sptlrpc_cli_unwrap_bulk_write);
2303
2304 /**
2305  * Performe transformation upon outgoing bulk read.
2306  */
2307 int sptlrpc_svc_wrap_bulk(struct ptlrpc_request *req,
2308                           struct ptlrpc_bulk_desc *desc)
2309 {
2310         struct ptlrpc_svc_ctx *ctx;
2311
2312         LASSERT(req->rq_bulk_read);
2313
2314         if (!req->rq_pack_bulk)
2315                 return 0;
2316
2317         ctx = req->rq_svc_ctx;
2318         if (ctx->sc_policy->sp_sops->wrap_bulk)
2319                 return ctx->sc_policy->sp_sops->wrap_bulk(req, desc);
2320
2321         return 0;
2322 }
2323 EXPORT_SYMBOL(sptlrpc_svc_wrap_bulk);
2324
2325 /**
2326  * Performe transformation upon incoming bulk write.
2327  */
2328 int sptlrpc_svc_unwrap_bulk(struct ptlrpc_request *req,
2329                             struct ptlrpc_bulk_desc *desc)
2330 {
2331         struct ptlrpc_svc_ctx *ctx;
2332         int                    rc;
2333
2334         LASSERT(req->rq_bulk_write);
2335
2336         /*
2337          * if it's in privacy mode, transferred should >= expected; otherwise
2338          * transferred should == expected.
2339          */
2340         if (desc->bd_nob_transferred < desc->bd_nob ||
2341             (desc->bd_nob_transferred > desc->bd_nob &&
2342              SPTLRPC_FLVR_BULK_SVC(req->rq_flvr.sf_rpc) !=
2343              SPTLRPC_BULK_SVC_PRIV)) {
2344                 DEBUG_REQ(D_ERROR, req, "truncated bulk GET %d(%d)",
2345                           desc->bd_nob_transferred, desc->bd_nob);
2346                 return -ETIMEDOUT;
2347         }
2348
2349         if (!req->rq_pack_bulk)
2350                 return 0;
2351
2352         ctx = req->rq_svc_ctx;
2353         if (ctx->sc_policy->sp_sops->unwrap_bulk) {
2354                 rc = ctx->sc_policy->sp_sops->unwrap_bulk(req, desc);
2355                 if (rc)
2356                         CERROR("error unwrap bulk: %d\n", rc);
2357         }
2358
2359         /* return 0 to allow reply be sent */
2360         return 0;
2361 }
2362 EXPORT_SYMBOL(sptlrpc_svc_unwrap_bulk);
2363
2364 /**
2365  * Prepare buffers for incoming bulk write.
2366  */
2367 int sptlrpc_svc_prep_bulk(struct ptlrpc_request *req,
2368                           struct ptlrpc_bulk_desc *desc)
2369 {
2370         struct ptlrpc_svc_ctx *ctx;
2371
2372         LASSERT(req->rq_bulk_write);
2373
2374         if (!req->rq_pack_bulk)
2375                 return 0;
2376
2377         ctx = req->rq_svc_ctx;
2378         if (ctx->sc_policy->sp_sops->prep_bulk)
2379                 return ctx->sc_policy->sp_sops->prep_bulk(req, desc);
2380
2381         return 0;
2382 }
2383 EXPORT_SYMBOL(sptlrpc_svc_prep_bulk);
2384
2385 /****************************************
2386  * user descriptor helpers              *
2387  ****************************************/
2388
2389 int sptlrpc_current_user_desc_size(void)
2390 {
2391         int ngroups;
2392
2393 #ifdef __KERNEL__
2394         ngroups = current_ngroups;
2395
2396         if (ngroups > LUSTRE_MAX_GROUPS)
2397                 ngroups = LUSTRE_MAX_GROUPS;
2398 #else
2399         ngroups = 0;
2400 #endif
2401         return sptlrpc_user_desc_size(ngroups);
2402 }
2403 EXPORT_SYMBOL(sptlrpc_current_user_desc_size);
2404
2405 int sptlrpc_pack_user_desc(struct lustre_msg *msg, int offset)
2406 {
2407         struct ptlrpc_user_desc *pud;
2408
2409         pud = lustre_msg_buf(msg, offset, 0);
2410
2411         pud->pud_uid = cfs_curproc_uid();
2412         pud->pud_gid = cfs_curproc_gid();
2413         pud->pud_fsuid = cfs_curproc_fsuid();
2414         pud->pud_fsgid = cfs_curproc_fsgid();
2415         pud->pud_cap = cfs_curproc_cap_pack();
2416         pud->pud_ngroups = (msg->lm_buflens[offset] - sizeof(*pud)) / 4;
2417
2418 #ifdef __KERNEL__
2419         task_lock(current);
2420         if (pud->pud_ngroups > current_ngroups)
2421                 pud->pud_ngroups = current_ngroups;
2422         memcpy(pud->pud_groups, current_cred()->group_info->blocks[0],
2423                pud->pud_ngroups * sizeof(__u32));
2424         task_unlock(current);
2425 #endif
2426
2427         return 0;
2428 }
2429 EXPORT_SYMBOL(sptlrpc_pack_user_desc);
2430
2431 int sptlrpc_unpack_user_desc(struct lustre_msg *msg, int offset, int swabbed)
2432 {
2433         struct ptlrpc_user_desc *pud;
2434         int                      i;
2435
2436         pud = lustre_msg_buf(msg, offset, sizeof(*pud));
2437         if (!pud)
2438                 return -EINVAL;
2439
2440         if (swabbed) {
2441                 __swab32s(&pud->pud_uid);
2442                 __swab32s(&pud->pud_gid);
2443                 __swab32s(&pud->pud_fsuid);
2444                 __swab32s(&pud->pud_fsgid);
2445                 __swab32s(&pud->pud_cap);
2446                 __swab32s(&pud->pud_ngroups);
2447         }
2448
2449         if (pud->pud_ngroups > LUSTRE_MAX_GROUPS) {
2450                 CERROR("%u groups is too large\n", pud->pud_ngroups);
2451                 return -EINVAL;
2452         }
2453
2454         if (sizeof(*pud) + pud->pud_ngroups * sizeof(__u32) >
2455             msg->lm_buflens[offset]) {
2456                 CERROR("%u groups are claimed but bufsize only %u\n",
2457                        pud->pud_ngroups, msg->lm_buflens[offset]);
2458                 return -EINVAL;
2459         }
2460
2461         if (swabbed) {
2462                 for (i = 0; i < pud->pud_ngroups; i++)
2463                         __swab32s(&pud->pud_groups[i]);
2464         }
2465
2466         return 0;
2467 }
2468 EXPORT_SYMBOL(sptlrpc_unpack_user_desc);
2469
2470 /****************************************
2471  * misc helpers                         *
2472  ****************************************/
2473
2474 const char * sec2target_str(struct ptlrpc_sec *sec)
2475 {
2476         if (!sec || !sec->ps_import || !sec->ps_import->imp_obd)
2477                 return "*";
2478         if (sec_is_reverse(sec))
2479                 return "c";
2480         return obd_uuid2str(&sec->ps_import->imp_obd->u.cli.cl_target_uuid);
2481 }
2482 EXPORT_SYMBOL(sec2target_str);
2483
2484 /*
2485  * return true if the bulk data is protected
2486  */
2487 int sptlrpc_flavor_has_bulk(struct sptlrpc_flavor *flvr)
2488 {
2489         switch (SPTLRPC_FLVR_BULK_SVC(flvr->sf_rpc)) {
2490         case SPTLRPC_BULK_SVC_INTG:
2491         case SPTLRPC_BULK_SVC_PRIV:
2492                 return 1;
2493         default:
2494                 return 0;
2495         }
2496 }
2497 EXPORT_SYMBOL(sptlrpc_flavor_has_bulk);
2498
2499 /****************************************
2500  * crypto API helper/alloc blkciper     *
2501  ****************************************/
2502
2503 /****************************************
2504  * initialize/finalize                  *
2505  ****************************************/
2506
2507 int sptlrpc_init(void)
2508 {
2509         int rc;
2510
2511         cfs_rwlock_init(&policy_lock);
2512
2513         rc = sptlrpc_gc_init();
2514         if (rc)
2515                 goto out;
2516
2517         rc = sptlrpc_conf_init();
2518         if (rc)
2519                 goto out_gc;
2520
2521         rc = sptlrpc_enc_pool_init();
2522         if (rc)
2523                 goto out_conf;
2524
2525         rc = sptlrpc_null_init();
2526         if (rc)
2527                 goto out_pool;
2528
2529         rc = sptlrpc_plain_init();
2530         if (rc)
2531                 goto out_null;
2532
2533         rc = sptlrpc_lproc_init();
2534         if (rc)
2535                 goto out_plain;
2536
2537         return 0;
2538
2539 out_plain:
2540         sptlrpc_plain_fini();
2541 out_null:
2542         sptlrpc_null_fini();
2543 out_pool:
2544         sptlrpc_enc_pool_fini();
2545 out_conf:
2546         sptlrpc_conf_fini();
2547 out_gc:
2548         sptlrpc_gc_fini();
2549 out:
2550         return rc;
2551 }
2552
2553 void sptlrpc_fini(void)
2554 {
2555         sptlrpc_lproc_fini();
2556         sptlrpc_plain_fini();
2557         sptlrpc_null_fini();
2558         sptlrpc_enc_pool_fini();
2559         sptlrpc_conf_fini();
2560         sptlrpc_gc_fini();
2561 }