kernel/drivers/staging/lustre/lustre/ptlrpc/sec.c

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