lustre/ptlrpc/sec.c

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