lustre/ptlrpc/gss/gss_pipefs.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Modifications for Lustre
   5  *
   6  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
   7  *
   8  * Author: Eric Mei <ericm@clusterfs.com>
   9  */
  10
  11 /*
  12  * linux/net/sunrpc/auth_gss.c
  13  *
  14  * RPCSEC_GSS client authentication.
  15  *
  16  *  Copyright (c) 2000 The Regents of the University of Michigan.
  17  *  All rights reserved.
  18  *
  19  *  Dug Song       <dugsong@monkey.org>
  20  *  Andy Adamson   <andros@umich.edu>
  21  *
  22  *  Redistribution and use in source and binary forms, with or without
  23  *  modification, are permitted provided that the following conditions
  24  *  are met:
  25  *
  26  *  1. Redistributions of source code must retain the above copyright
  27  *     notice, this list of conditions and the following disclaimer.
  28  *  2. Redistributions in binary form must reproduce the above copyright
  29  *     notice, this list of conditions and the following disclaimer in the
  30  *     documentation and/or other materials provided with the distribution.
  31  *  3. Neither the name of the University nor the names of its
  32  *     contributors may be used to endorse or promote products derived
  33  *     from this software without specific prior written permission.
  34  *
  35  *  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESS OR IMPLIED
  36  *  WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
  37  *  MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  38  *  DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
  39  *  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  40  *  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  41  *  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
  42  *  BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
  43  *  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
  44  *  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  45  *  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  46  *
  47  */
  48
  49 #ifndef EXPORT_SYMTAB
  50 # define EXPORT_SYMTAB
  51 #endif
  52 #define DEBUG_SUBSYSTEM S_SEC
  53 #ifdef __KERNEL__
  54 #include <linux/init.h>
  55 #include <linux/module.h>
  56 #include <linux/slab.h>
  57 #include <linux/dcache.h>
  58 #include <linux/fs.h>
  59 #include <linux/random.h>
  60 #include <linux/mutex.h>
  61 #include <linux/crypto.h>
  62 #include <asm/atomic.h>
  63 struct rpc_clnt; /* for rpc_pipefs */
  64 #include <linux/sunrpc/rpc_pipe_fs.h>
  65 #else
  66 #include <liblustre.h>
  67 #endif
  68
  69 #include <obd.h>
  70 #include <obd_class.h>
  71 #include <obd_support.h>
  72 #include <lustre/lustre_idl.h>
  73 #include <lustre_sec.h>
  74 #include <lustre_net.h>
  75 #include <lustre_import.h>
  76
  77 #include "gss_err.h"
  78 #include "gss_internal.h"
  79 #include "gss_api.h"
  80
  81 static struct ptlrpc_sec_policy gss_policy_pipefs;
  82 static struct ptlrpc_ctx_ops gss_pipefs_ctxops;
  83
  84 static int gss_cli_ctx_refresh_pf(struct ptlrpc_cli_ctx *ctx);
  85
  86 static int gss_sec_pipe_upcall_init(struct gss_sec *gsec)
  87 {
  88         return 0;
  89 }
  90
  91 static void gss_sec_pipe_upcall_fini(struct gss_sec *gsec)
  92 {
  93 }
  94
  95 /****************************************
  96  * internel context helpers             *
  97  ****************************************/
  98
  99 static
 100 struct ptlrpc_cli_ctx *ctx_create_pf(struct ptlrpc_sec *sec,
 101                                      struct vfs_cred *vcred)
 102 {
 103         struct gss_cli_ctx *gctx;
 104         int                 rc;
 105
 106         OBD_ALLOC_PTR(gctx);
 107         if (gctx == NULL)
 108                 return NULL;
 109
 110         rc = gss_cli_ctx_init_common(sec, &gctx->gc_base,
 111                                      &gss_pipefs_ctxops, vcred);
 112         if (rc) {
 113                 OBD_FREE_PTR(gctx);
 114                 return NULL;
 115         }
 116
 117         return &gctx->gc_base;
 118 }
 119
 120 static
 121 void ctx_destroy_pf(struct ptlrpc_sec *sec, struct ptlrpc_cli_ctx *ctx)
 122 {
 123         struct gss_cli_ctx *gctx = ctx2gctx(ctx);
 124
 125         if (gss_cli_ctx_fini_common(sec, ctx))
 126                 return;
 127
 128         OBD_FREE_PTR(gctx);
 129
 130         cfs_atomic_dec(&sec->ps_nctx);
 131         sptlrpc_sec_put(sec);
 132 }
 133
 134 static
 135 void ctx_enhash_pf(struct ptlrpc_cli_ctx *ctx, cfs_hlist_head_t *hash)
 136 {
 137         cfs_set_bit(PTLRPC_CTX_CACHED_BIT, &ctx->cc_flags);
 138         cfs_atomic_inc(&ctx->cc_refcount);
 139         cfs_hlist_add_head(&ctx->cc_cache, hash);
 140 }
 141
 142 /*
 143  * caller must hold spinlock
 144  */
 145 static
 146 void ctx_unhash_pf(struct ptlrpc_cli_ctx *ctx, cfs_hlist_head_t *freelist)
 147 {
 148         LASSERT_SPIN_LOCKED(&ctx->cc_sec->ps_lock);
 149         LASSERT(cfs_atomic_read(&ctx->cc_refcount) > 0);
 150         LASSERT(cfs_test_bit(PTLRPC_CTX_CACHED_BIT, &ctx->cc_flags));
 151         LASSERT(!cfs_hlist_unhashed(&ctx->cc_cache));
 152
 153         cfs_clear_bit(PTLRPC_CTX_CACHED_BIT, &ctx->cc_flags);
 154
 155         if (cfs_atomic_dec_and_test(&ctx->cc_refcount)) {
 156                 __cfs_hlist_del(&ctx->cc_cache);
 157                 cfs_hlist_add_head(&ctx->cc_cache, freelist);
 158         } else {
 159                 cfs_hlist_del_init(&ctx->cc_cache);
 160         }
 161 }
 162
 163 /*
 164  * return 1 if the context is dead.
 165  */
 166 static
 167 int ctx_check_death_pf(struct ptlrpc_cli_ctx *ctx,
 168                        cfs_hlist_head_t *freelist)
 169 {
 170         if (cli_ctx_check_death(ctx)) {
 171                 if (freelist)
 172                         ctx_unhash_pf(ctx, freelist);
 173                 return 1;
 174         }
 175
 176         return 0;
 177 }
 178
 179 static inline
 180 int ctx_check_death_locked_pf(struct ptlrpc_cli_ctx *ctx,
 181                               cfs_hlist_head_t *freelist)
 182 {
 183         LASSERT(ctx->cc_sec);
 184         LASSERT(cfs_atomic_read(&ctx->cc_refcount) > 0);
 185         LASSERT(cfs_test_bit(PTLRPC_CTX_CACHED_BIT, &ctx->cc_flags));
 186
 187         return ctx_check_death_pf(ctx, freelist);
 188 }
 189
 190 static inline
 191 int ctx_match_pf(struct ptlrpc_cli_ctx *ctx, struct vfs_cred *vcred)
 192 {
 193         /* a little bit optimization for null policy */
 194         if (!ctx->cc_ops->match)
 195                 return 1;
 196
 197         return ctx->cc_ops->match(ctx, vcred);
 198 }
 199
 200 static
 201 void ctx_list_destroy_pf(cfs_hlist_head_t *head)
 202 {
 203         struct ptlrpc_cli_ctx *ctx;
 204
 205         while (!cfs_hlist_empty(head)) {
 206                 ctx = cfs_hlist_entry(head->first, struct ptlrpc_cli_ctx,
 207                                       cc_cache);
 208
 209                 LASSERT(cfs_atomic_read(&ctx->cc_refcount) == 0);
 210                 LASSERT(cfs_test_bit(PTLRPC_CTX_CACHED_BIT,
 211                                      &ctx->cc_flags) == 0);
 212
 213                 cfs_hlist_del_init(&ctx->cc_cache);
 214                 ctx_destroy_pf(ctx->cc_sec, ctx);
 215         }
 216 }
 217
 218 /****************************************
 219  * context apis                         *
 220  ****************************************/
 221
 222 static
 223 int gss_cli_ctx_validate_pf(struct ptlrpc_cli_ctx *ctx)
 224 {
 225         if (ctx_check_death_pf(ctx, NULL))
 226                 return 1;
 227         if (cli_ctx_is_ready(ctx))
 228                 return 0;
 229         return 1;
 230 }
 231
 232 static
 233 void gss_cli_ctx_die_pf(struct ptlrpc_cli_ctx *ctx, int grace)
 234 {
 235         LASSERT(ctx->cc_sec);
 236         LASSERT(cfs_atomic_read(&ctx->cc_refcount) > 0);
 237
 238         cli_ctx_expire(ctx);
 239
 240         cfs_spin_lock(&ctx->cc_sec->ps_lock);
 241
 242         if (cfs_test_and_clear_bit(PTLRPC_CTX_CACHED_BIT, &ctx->cc_flags)) {
 243                 LASSERT(!cfs_hlist_unhashed(&ctx->cc_cache));
 244                 LASSERT(cfs_atomic_read(&ctx->cc_refcount) > 1);
 245
 246                 cfs_hlist_del_init(&ctx->cc_cache);
 247                 if (cfs_atomic_dec_and_test(&ctx->cc_refcount))
 248                         LBUG();
 249         }
 250
 251         cfs_spin_unlock(&ctx->cc_sec->ps_lock);
 252 }
 253
 254 /****************************************
 255  * reverse context installation         *
 256  ****************************************/
 257
 258 static inline
 259 unsigned int ctx_hash_index(int hashsize, __u64 key)
 260 {
 261         return (unsigned int) (key & ((__u64) hashsize - 1));
 262 }
 263
 264 static
 265 void gss_sec_ctx_replace_pf(struct gss_sec *gsec,
 266                             struct ptlrpc_cli_ctx *new)
 267 {
 268         struct gss_sec_pipefs *gsec_pf;
 269         struct ptlrpc_cli_ctx *ctx;
 270         cfs_hlist_node_t      *pos, *next;
 271         CFS_HLIST_HEAD(freelist);
 272         unsigned int hash;
 273         ENTRY;
 274
 275         gsec_pf = container_of(gsec, struct gss_sec_pipefs, gsp_base);
 276
 277         hash = ctx_hash_index(gsec_pf->gsp_chash_size,
 278                               (__u64) new->cc_vcred.vc_uid);
 279         LASSERT(hash < gsec_pf->gsp_chash_size);
 280
 281         cfs_spin_lock(&gsec->gs_base.ps_lock);
 282
 283         cfs_hlist_for_each_entry_safe(ctx, pos, next,
 284                                       &gsec_pf->gsp_chash[hash], cc_cache) {
 285                 if (!ctx_match_pf(ctx, &new->cc_vcred))
 286                         continue;
 287
 288                 cli_ctx_expire(ctx);
 289                 ctx_unhash_pf(ctx, &freelist);
 290                 break;
 291         }
 292
 293         ctx_enhash_pf(new, &gsec_pf->gsp_chash[hash]);
 294
 295         cfs_spin_unlock(&gsec->gs_base.ps_lock);
 296
 297         ctx_list_destroy_pf(&freelist);
 298         EXIT;
 299 }
 300
 301 static
 302 int gss_install_rvs_cli_ctx_pf(struct gss_sec *gsec,
 303                                struct ptlrpc_svc_ctx *svc_ctx)
 304 {
 305         struct vfs_cred          vcred;
 306         struct ptlrpc_cli_ctx   *cli_ctx;
 307         int                      rc;
 308         ENTRY;
 309
 310         vcred.vc_uid = 0;
 311         vcred.vc_gid = 0;
 312
 313         cli_ctx = ctx_create_pf(&gsec->gs_base, &vcred);
 314         if (!cli_ctx)
 315                 RETURN(-ENOMEM);
 316
 317         rc = gss_copy_rvc_cli_ctx(cli_ctx, svc_ctx);
 318         if (rc) {
 319                 ctx_destroy_pf(cli_ctx->cc_sec, cli_ctx);
 320                 RETURN(rc);
 321         }
 322
 323         gss_sec_ctx_replace_pf(gsec, cli_ctx);
 324         RETURN(0);
 325 }
 326
 327 static
 328 void gss_ctx_cache_gc_pf(struct gss_sec_pipefs *gsec_pf,
 329                          cfs_hlist_head_t *freelist)
 330 {
 331         struct ptlrpc_sec       *sec;
 332         struct ptlrpc_cli_ctx   *ctx;
 333         cfs_hlist_node_t        *pos, *next;
 334         int i;
 335         ENTRY;
 336
 337         sec = &gsec_pf->gsp_base.gs_base;
 338
 339         CDEBUG(D_SEC, "do gc on sec %s@%p\n", sec->ps_policy->sp_name, sec);
 340
 341         for (i = 0; i < gsec_pf->gsp_chash_size; i++) {
 342                 cfs_hlist_for_each_entry_safe(ctx, pos, next,
 343                                               &gsec_pf->gsp_chash[i], cc_cache)
 344                         ctx_check_death_locked_pf(ctx, freelist);
 345         }
 346
 347         sec->ps_gc_next = cfs_time_current_sec() + sec->ps_gc_interval;
 348         EXIT;
 349 }
 350
 351 static
 352 struct ptlrpc_sec* gss_sec_create_pf(struct obd_import *imp,
 353                                      struct ptlrpc_svc_ctx *ctx,
 354                                      struct sptlrpc_flavor *sf)
 355 {
 356         struct gss_sec_pipefs   *gsec_pf;
 357         int                      alloc_size, hash_size, i;
 358         ENTRY;
 359
 360 #define GSS_SEC_PIPEFS_CTX_HASH_SIZE    (32)
 361
 362         if (ctx ||
 363             sf->sf_flags & (PTLRPC_SEC_FL_ROOTONLY | PTLRPC_SEC_FL_REVERSE))
 364                 hash_size = 1;
 365         else
 366                 hash_size = GSS_SEC_PIPEFS_CTX_HASH_SIZE;
 367
 368         alloc_size = sizeof(*gsec_pf) +
 369                      sizeof(cfs_hlist_head_t) * hash_size;
 370
 371         OBD_ALLOC(gsec_pf, alloc_size);
 372         if (!gsec_pf)
 373                 RETURN(NULL);
 374
 375         gsec_pf->gsp_chash_size = hash_size;
 376         for (i = 0; i < hash_size; i++)
 377                 CFS_INIT_HLIST_HEAD(&gsec_pf->gsp_chash[i]);
 378
 379         if (gss_sec_create_common(&gsec_pf->gsp_base, &gss_policy_pipefs,
 380                                   imp, ctx, sf))
 381                 goto err_free;
 382
 383         if (ctx == NULL) {
 384                 if (gss_sec_pipe_upcall_init(&gsec_pf->gsp_base))
 385                         goto err_destroy;
 386         } else {
 387                 if (gss_install_rvs_cli_ctx_pf(&gsec_pf->gsp_base, ctx))
 388                         goto err_destroy;
 389         }
 390
 391         RETURN(&gsec_pf->gsp_base.gs_base);
 392
 393 err_destroy:
 394         gss_sec_destroy_common(&gsec_pf->gsp_base);
 395 err_free:
 396         OBD_FREE(gsec_pf, alloc_size);
 397         RETURN(NULL);
 398 }
 399
 400 static
 401 void gss_sec_destroy_pf(struct ptlrpc_sec *sec)
 402 {
 403         struct gss_sec_pipefs   *gsec_pf;
 404         struct gss_sec          *gsec;
 405
 406         CWARN("destroy %s@%p\n", sec->ps_policy->sp_name, sec);
 407
 408         gsec = container_of(sec, struct gss_sec, gs_base);
 409         gsec_pf = container_of(gsec, struct gss_sec_pipefs, gsp_base);
 410
 411         LASSERT(gsec_pf->gsp_chash);
 412         LASSERT(gsec_pf->gsp_chash_size);
 413
 414         gss_sec_pipe_upcall_fini(gsec);
 415
 416         gss_sec_destroy_common(gsec);
 417
 418         OBD_FREE(gsec, sizeof(*gsec_pf) +
 419                        sizeof(cfs_hlist_head_t) * gsec_pf->gsp_chash_size);
 420 }
 421
 422 static
 423 struct ptlrpc_cli_ctx * gss_sec_lookup_ctx_pf(struct ptlrpc_sec *sec,
 424                                               struct vfs_cred *vcred,
 425                                               int create, int remove_dead)
 426 {
 427         struct gss_sec         *gsec;
 428         struct gss_sec_pipefs  *gsec_pf;
 429         struct ptlrpc_cli_ctx  *ctx = NULL, *new = NULL;
 430         cfs_hlist_head_t       *hash_head;
 431         cfs_hlist_node_t       *pos, *next;
 432         CFS_HLIST_HEAD(freelist);
 433         unsigned int            hash, gc = 0, found = 0;
 434         ENTRY;
 435
 436         cfs_might_sleep();
 437
 438         gsec = container_of(sec, struct gss_sec, gs_base);
 439         gsec_pf = container_of(gsec, struct gss_sec_pipefs, gsp_base);
 440
 441         hash = ctx_hash_index(gsec_pf->gsp_chash_size,
 442                               (__u64) vcred->vc_uid);
 443         hash_head = &gsec_pf->gsp_chash[hash];
 444         LASSERT(hash < gsec_pf->gsp_chash_size);
 445
 446 retry:
 447         cfs_spin_lock(&sec->ps_lock);
 448
 449         /* gc_next == 0 means never do gc */
 450         if (remove_dead && sec->ps_gc_next &&
 451             cfs_time_after(cfs_time_current_sec(), sec->ps_gc_next)) {
 452                 gss_ctx_cache_gc_pf(gsec_pf, &freelist);
 453                 gc = 1;
 454         }
 455
 456         cfs_hlist_for_each_entry_safe(ctx, pos, next, hash_head, cc_cache) {
 457                 if (gc == 0 &&
 458                     ctx_check_death_locked_pf(ctx,
 459                                               remove_dead ? &freelist : NULL))
 460                         continue;
 461
 462                 if (ctx_match_pf(ctx, vcred)) {
 463                         found = 1;
 464                         break;
 465                 }
 466         }
 467
 468         if (found) {
 469                 if (new && new != ctx) {
 470                         /* lost the race, just free it */
 471                         cfs_hlist_add_head(&new->cc_cache, &freelist);
 472                         new = NULL;
 473                 }
 474
 475                 /* hot node, move to head */
 476                 if (hash_head->first != &ctx->cc_cache) {
 477                         __cfs_hlist_del(&ctx->cc_cache);
 478                         cfs_hlist_add_head(&ctx->cc_cache, hash_head);
 479                 }
 480         } else {
 481                 /* don't allocate for reverse sec */
 482                 if (sec_is_reverse(sec)) {
 483                         cfs_spin_unlock(&sec->ps_lock);
 484                         RETURN(NULL);
 485                 }
 486
 487                 if (new) {
 488                         ctx_enhash_pf(new, hash_head);
 489                         ctx = new;
 490                 } else if (create) {
 491                         cfs_spin_unlock(&sec->ps_lock);
 492                         new = ctx_create_pf(sec, vcred);
 493                         if (new) {
 494                                 cfs_clear_bit(PTLRPC_CTX_NEW_BIT,
 495                                               &new->cc_flags);
 496                                 goto retry;
 497                         }
 498                 } else
 499                         ctx = NULL;
 500         }
 501
 502         /* hold a ref */
 503         if (ctx)
 504                 cfs_atomic_inc(&ctx->cc_refcount);
 505
 506         cfs_spin_unlock(&sec->ps_lock);
 507
 508         /* the allocator of the context must give the first push to refresh */
 509         if (new) {
 510                 LASSERT(new == ctx);
 511                 gss_cli_ctx_refresh_pf(new);
 512         }
 513
 514         ctx_list_destroy_pf(&freelist);
 515         RETURN(ctx);
 516 }
 517
 518 static
 519 void gss_sec_release_ctx_pf(struct ptlrpc_sec *sec,
 520                             struct ptlrpc_cli_ctx *ctx,
 521                             int sync)
 522 {
 523         LASSERT(cfs_test_bit(PTLRPC_CTX_CACHED_BIT, &ctx->cc_flags) == 0);
 524         LASSERT(cfs_hlist_unhashed(&ctx->cc_cache));
 525
 526         /* if required async, we must clear the UPTODATE bit to prevent extra
 527          * rpcs during destroy procedure. */
 528         if (!sync)
 529                 cfs_clear_bit(PTLRPC_CTX_UPTODATE_BIT, &ctx->cc_flags);
 530
 531         /* destroy this context */
 532         ctx_destroy_pf(sec, ctx);
 533 }
 534
 535 /*
 536  * @uid: which user. "-1" means flush all.
 537  * @grace: mark context DEAD, allow graceful destroy like notify
 538  *         server side, etc.
 539  * @force: also flush busy entries.
 540  *
 541  * return the number of busy context encountered.
 542  *
 543  * In any cases, never touch "eternal" contexts.
 544  */
 545 static
 546 int gss_sec_flush_ctx_cache_pf(struct ptlrpc_sec *sec,
 547                                uid_t uid,
 548                                int grace, int force)
 549 {
 550         struct gss_sec          *gsec;
 551         struct gss_sec_pipefs   *gsec_pf;
 552         struct ptlrpc_cli_ctx   *ctx;
 553         cfs_hlist_node_t        *pos, *next;
 554         CFS_HLIST_HEAD(freelist);
 555         int i, busy = 0;
 556         ENTRY;
 557
 558         might_sleep_if(grace);
 559
 560         gsec = container_of(sec, struct gss_sec, gs_base);
 561         gsec_pf = container_of(gsec, struct gss_sec_pipefs, gsp_base);
 562
 563         cfs_spin_lock(&sec->ps_lock);
 564         for (i = 0; i < gsec_pf->gsp_chash_size; i++) {
 565                 cfs_hlist_for_each_entry_safe(ctx, pos, next,
 566                                               &gsec_pf->gsp_chash[i],
 567                                               cc_cache) {
 568                         LASSERT(cfs_atomic_read(&ctx->cc_refcount) > 0);
 569
 570                         if (uid != -1 && uid != ctx->cc_vcred.vc_uid)
 571                                 continue;
 572
 573                         if (cfs_atomic_read(&ctx->cc_refcount) > 1) {
 574                                 busy++;
 575                                 if (!force)
 576                                         continue;
 577
 578                                 CWARN("flush busy(%d) ctx %p(%u->%s) by force, "
 579                                       "grace %d\n",
 580                                       cfs_atomic_read(&ctx->cc_refcount),
 581                                       ctx, ctx->cc_vcred.vc_uid,
 582                                       sec2target_str(ctx->cc_sec), grace);
 583                         }
 584                         ctx_unhash_pf(ctx, &freelist);
 585
 586                         cfs_set_bit(PTLRPC_CTX_DEAD_BIT, &ctx->cc_flags);
 587                         if (!grace)
 588                                 cfs_clear_bit(PTLRPC_CTX_UPTODATE_BIT,
 589                                               &ctx->cc_flags);
 590                 }
 591         }
 592         cfs_spin_unlock(&sec->ps_lock);
 593
 594         ctx_list_destroy_pf(&freelist);
 595         RETURN(busy);
 596 }
 597
 598 /****************************************
 599  * service apis                         *
 600  ****************************************/
 601
 602 static
 603 int gss_svc_accept_pf(struct ptlrpc_request *req)
 604 {
 605         return gss_svc_accept(&gss_policy_pipefs, req);
 606 }
 607
 608 static
 609 int gss_svc_install_rctx_pf(struct obd_import *imp,
 610                             struct ptlrpc_svc_ctx *ctx)
 611 {
 612         struct ptlrpc_sec *sec;
 613         int                rc;
 614
 615         sec = sptlrpc_import_sec_ref(imp);
 616         LASSERT(sec);
 617         rc = gss_install_rvs_cli_ctx_pf(sec2gsec(sec), ctx);
 618
 619         sptlrpc_sec_put(sec);
 620         return rc;
 621 }
 622
 623 /****************************************
 624  * rpc_pipefs definitions               *
 625  ****************************************/
 626
 627 #define LUSTRE_PIPE_ROOT        "/lustre"
 628 #define LUSTRE_PIPE_KRB5        LUSTRE_PIPE_ROOT"/krb5"
 629
 630 struct gss_upcall_msg_data {
 631         __u32                           gum_seq;
 632         __u32                           gum_uid;
 633         __u32                           gum_gid;
 634         __u32                           gum_svc;        /* MDS/OSS... */
 635         __u64                           gum_nid;        /* peer NID */
 636         __u8                            gum_obd[64];    /* client obd name */
 637 };
 638
 639 struct gss_upcall_msg {
 640         struct rpc_pipe_msg             gum_base;
 641         cfs_atomic_t                    gum_refcount;
 642         cfs_list_t                      gum_list;
 643         __u32                           gum_mechidx;
 644         struct gss_sec                 *gum_gsec;
 645         struct gss_cli_ctx             *gum_gctx;
 646         struct gss_upcall_msg_data      gum_data;
 647 };
 648
 649 static cfs_atomic_t upcall_seq = CFS_ATOMIC_INIT(0);
 650
 651 static inline
 652 __u32 upcall_get_sequence(void)
 653 {
 654         return (__u32) cfs_atomic_inc_return(&upcall_seq);
 655 }
 656
 657 enum mech_idx_t {
 658         MECH_KRB5   = 0,
 659         MECH_MAX
 660 };
 661
 662 static inline
 663 __u32 mech_name2idx(const char *name)
 664 {
 665         LASSERT(!strcmp(name, "krb5"));
 666         return MECH_KRB5;
 667 }
 668
 669 /* pipefs dentries for each mechanisms */
 670 static struct dentry *de_pipes[MECH_MAX] = { NULL, };
 671 /* all upcall messgaes linked here */
 672 static cfs_list_t upcall_lists[MECH_MAX];
 673 /* and protected by this */
 674 static cfs_spinlock_t upcall_locks[MECH_MAX];
 675
 676 static inline
 677 void upcall_list_lock(int idx)
 678 {
 679         cfs_spin_lock(&upcall_locks[idx]);
 680 }
 681
 682 static inline
 683 void upcall_list_unlock(int idx)
 684 {
 685         cfs_spin_unlock(&upcall_locks[idx]);
 686 }
 687
 688 static
 689 void upcall_msg_enlist(struct gss_upcall_msg *msg)
 690 {
 691         __u32 idx = msg->gum_mechidx;
 692
 693         upcall_list_lock(idx);
 694         cfs_list_add(&msg->gum_list, &upcall_lists[idx]);
 695         upcall_list_unlock(idx);
 696 }
 697
 698 static
 699 void upcall_msg_delist(struct gss_upcall_msg *msg)
 700 {
 701         __u32 idx = msg->gum_mechidx;
 702
 703         upcall_list_lock(idx);
 704         cfs_list_del_init(&msg->gum_list);
 705         upcall_list_unlock(idx);
 706 }
 707
 708 /****************************************
 709  * rpc_pipefs upcall helpers            *
 710  ****************************************/
 711
 712 static
 713 void gss_release_msg(struct gss_upcall_msg *gmsg)
 714 {
 715         ENTRY;
 716         LASSERT(cfs_atomic_read(&gmsg->gum_refcount) > 0);
 717
 718         if (!cfs_atomic_dec_and_test(&gmsg->gum_refcount)) {
 719                 EXIT;
 720                 return;
 721         }
 722
 723         if (gmsg->gum_gctx) {
 724                 sptlrpc_cli_ctx_wakeup(&gmsg->gum_gctx->gc_base);
 725                 sptlrpc_cli_ctx_put(&gmsg->gum_gctx->gc_base, 1);
 726                 gmsg->gum_gctx = NULL;
 727         }
 728
 729         LASSERT(cfs_list_empty(&gmsg->gum_list));
 730         LASSERT(cfs_list_empty(&gmsg->gum_base.list));
 731         OBD_FREE_PTR(gmsg);
 732         EXIT;
 733 }
 734
 735 static
 736 void gss_unhash_msg_nolock(struct gss_upcall_msg *gmsg)
 737 {
 738         __u32 idx = gmsg->gum_mechidx;
 739
 740         LASSERT(idx < MECH_MAX);
 741         LASSERT_SPIN_LOCKED(&upcall_locks[idx]);
 742
 743         if (cfs_list_empty(&gmsg->gum_list))
 744                 return;
 745
 746         cfs_list_del_init(&gmsg->gum_list);
 747         LASSERT(cfs_atomic_read(&gmsg->gum_refcount) > 1);
 748         cfs_atomic_dec(&gmsg->gum_refcount);
 749 }
 750
 751 static
 752 void gss_unhash_msg(struct gss_upcall_msg *gmsg)
 753 {
 754         __u32 idx = gmsg->gum_mechidx;
 755
 756         LASSERT(idx < MECH_MAX);
 757         upcall_list_lock(idx);
 758         gss_unhash_msg_nolock(gmsg);
 759         upcall_list_unlock(idx);
 760 }
 761
 762 static
 763 void gss_msg_fail_ctx(struct gss_upcall_msg *gmsg)
 764 {
 765         if (gmsg->gum_gctx) {
 766                 struct ptlrpc_cli_ctx *ctx = &gmsg->gum_gctx->gc_base;
 767
 768                 LASSERT(cfs_atomic_read(&ctx->cc_refcount) > 0);
 769                 sptlrpc_cli_ctx_expire(ctx);
 770                 cfs_set_bit(PTLRPC_CTX_ERROR_BIT, &ctx->cc_flags);
 771         }
 772 }
 773
 774 static
 775 struct gss_upcall_msg * gss_find_upcall(__u32 mechidx, __u32 seq)
 776 {
 777         struct gss_upcall_msg *gmsg;
 778
 779         upcall_list_lock(mechidx);
 780         cfs_list_for_each_entry(gmsg, &upcall_lists[mechidx], gum_list) {
 781                 if (gmsg->gum_data.gum_seq != seq)
 782                         continue;
 783
 784                 LASSERT(cfs_atomic_read(&gmsg->gum_refcount) > 0);
 785                 LASSERT(gmsg->gum_mechidx == mechidx);
 786
 787                 cfs_atomic_inc(&gmsg->gum_refcount);
 788                 upcall_list_unlock(mechidx);
 789                 return gmsg;
 790         }
 791         upcall_list_unlock(mechidx);
 792         return NULL;
 793 }
 794
 795 static
 796 int simple_get_bytes(char **buf, __u32 *buflen, void *res, __u32 reslen)
 797 {
 798         if (*buflen < reslen) {
 799                 CERROR("buflen %u < %u\n", *buflen, reslen);
 800                 return -EINVAL;
 801         }
 802
 803         memcpy(res, *buf, reslen);
 804         *buf += reslen;
 805         *buflen -= reslen;
 806         return 0;
 807 }
 808
 809 /****************************************
 810  * rpc_pipefs apis                      *
 811  ****************************************/
 812
 813 static
 814 ssize_t gss_pipe_upcall(struct file *filp, struct rpc_pipe_msg *msg,
 815                         char *dst, size_t buflen)
 816 {
 817         char *data = (char *)msg->data + msg->copied;
 818         ssize_t mlen = msg->len;
 819         ssize_t left;
 820         ENTRY;
 821
 822         if (mlen > buflen)
 823                 mlen = buflen;
 824         left = cfs_copy_to_user(dst, data, mlen);
 825         if (left < 0) {
 826                 msg->errno = left;
 827                 RETURN(left);
 828         }
 829         mlen -= left;
 830         msg->copied += mlen;
 831         msg->errno = 0;
 832         RETURN(mlen);
 833 }
 834
 835 static
 836 ssize_t gss_pipe_downcall(struct file *filp, const char *src, size_t mlen)
 837 {
 838         struct rpc_inode        *rpci = RPC_I(filp->f_dentry->d_inode);
 839         struct gss_upcall_msg   *gss_msg;
 840         struct ptlrpc_cli_ctx   *ctx;
 841         struct gss_cli_ctx      *gctx = NULL;
 842         char                    *buf, *data;
 843         int                      datalen;
 844         int                      timeout, rc;
 845         __u32                    mechidx, seq, gss_err;
 846         ENTRY;
 847
 848         mechidx = (__u32) (long) rpci->private;
 849         LASSERT(mechidx < MECH_MAX);
 850
 851         OBD_ALLOC(buf, mlen);
 852         if (!buf)
 853                 RETURN(-ENOMEM);
 854
 855         if (cfs_copy_from_user(buf, src, mlen)) {
 856                 CERROR("failed copy user space data\n");
 857                 GOTO(out_free, rc = -EFAULT);
 858         }
 859         data = buf;
 860         datalen = mlen;
 861
 862         /* data passed down format:
 863          *  - seq
 864          *  - timeout
 865          *  - gc_win / error
 866          *  - wire_ctx (rawobj)
 867          *  - mech_ctx (rawobj)
 868          */
 869         if (simple_get_bytes(&data, &datalen, &seq, sizeof(seq))) {
 870                 CERROR("fail to get seq\n");
 871                 GOTO(out_free, rc = -EFAULT);
 872         }
 873
 874         gss_msg = gss_find_upcall(mechidx, seq);
 875         if (!gss_msg) {
 876                 CERROR("upcall %u has aborted earlier\n", seq);
 877                 GOTO(out_free, rc = -EINVAL);
 878         }
 879
 880         gss_unhash_msg(gss_msg);
 881         gctx = gss_msg->gum_gctx;
 882         LASSERT(gctx);
 883         LASSERT(cfs_atomic_read(&gctx->gc_base.cc_refcount) > 0);
 884
 885         /* timeout is not in use for now */
 886         if (simple_get_bytes(&data, &datalen, &timeout, sizeof(timeout)))
 887                 GOTO(out_msg, rc = -EFAULT);
 888
 889         /* lgssd signal an error by gc_win == 0 */
 890         if (simple_get_bytes(&data, &datalen, &gctx->gc_win,
 891                              sizeof(gctx->gc_win)))
 892                 GOTO(out_msg, rc = -EFAULT);
 893
 894         if (gctx->gc_win == 0) {
 895                 /* followed by:
 896                  * - rpc error
 897                  * - gss error
 898                  */
 899                 if (simple_get_bytes(&data, &datalen, &rc, sizeof(rc)))
 900                         GOTO(out_msg, rc = -EFAULT);
 901                 if (simple_get_bytes(&data, &datalen, &gss_err,sizeof(gss_err)))
 902                         GOTO(out_msg, rc = -EFAULT);
 903
 904                 if (rc == 0 && gss_err == GSS_S_COMPLETE) {
 905                         CWARN("both rpc & gss error code not set\n");
 906                         rc = -EPERM;
 907                 }
 908         } else {
 909                 rawobj_t tmpobj;
 910
 911                 /* handle */
 912                 if (rawobj_extract_local(&tmpobj, (__u32 **) &data, &datalen))
 913                         GOTO(out_msg, rc = -EFAULT);
 914                 if (rawobj_dup(&gctx->gc_handle, &tmpobj))
 915                         GOTO(out_msg, rc = -ENOMEM);
 916
 917                 /* mechctx */
 918                 if (rawobj_extract_local(&tmpobj, (__u32 **) &data, &datalen))
 919                         GOTO(out_msg, rc = -EFAULT);
 920                 gss_err = lgss_import_sec_context(&tmpobj,
 921                                                   gss_msg->gum_gsec->gs_mech,
 922                                                   &gctx->gc_mechctx);
 923                 rc = 0;
 924         }
 925
 926         if (likely(rc == 0 && gss_err == GSS_S_COMPLETE)) {
 927                 gss_cli_ctx_uptodate(gctx);
 928         } else {
 929                 ctx = &gctx->gc_base;
 930                 sptlrpc_cli_ctx_expire(ctx);
 931                 if (rc != -ERESTART || gss_err != GSS_S_COMPLETE)
 932                         cfs_set_bit(PTLRPC_CTX_ERROR_BIT, &ctx->cc_flags);
 933
 934                 CERROR("refresh ctx %p(uid %d) failed: %d/0x%08x: %s\n",
 935                        ctx, ctx->cc_vcred.vc_uid, rc, gss_err,
 936                        cfs_test_bit(PTLRPC_CTX_ERROR_BIT, &ctx->cc_flags) ?
 937                        "fatal error" : "non-fatal");
 938         }
 939
 940         rc = mlen;
 941
 942 out_msg:
 943         gss_release_msg(gss_msg);
 944
 945 out_free:
 946         OBD_FREE(buf, mlen);
 947         /* FIXME
 948          * hack pipefs: always return asked length unless all following
 949          * downcalls might be messed up. */
 950         rc = mlen;
 951         RETURN(rc);
 952 }
 953
 954 static
 955 void gss_pipe_destroy_msg(struct rpc_pipe_msg *msg)
 956 {
 957         struct gss_upcall_msg          *gmsg;
 958         struct gss_upcall_msg_data     *gumd;
 959         static cfs_time_t               ratelimit = 0;
 960         ENTRY;
 961
 962         LASSERT(cfs_list_empty(&msg->list));
 963
 964         /* normally errno is >= 0 */
 965         if (msg->errno >= 0) {
 966                 EXIT;
 967                 return;
 968         }
 969
 970         gmsg = container_of(msg, struct gss_upcall_msg, gum_base);
 971         gumd = &gmsg->gum_data;
 972         LASSERT(cfs_atomic_read(&gmsg->gum_refcount) > 0);
 973
 974         CERROR("failed msg %p (seq %u, uid %u, svc %u, nid "LPX64", obd %.*s): "
 975                "errno %d\n", msg, gumd->gum_seq, gumd->gum_uid, gumd->gum_svc,
 976                gumd->gum_nid, (int) sizeof(gumd->gum_obd),
 977                gumd->gum_obd, msg->errno);
 978
 979         cfs_atomic_inc(&gmsg->gum_refcount);
 980         gss_unhash_msg(gmsg);
 981         if (msg->errno == -ETIMEDOUT || msg->errno == -EPIPE) {
 982                 cfs_time_t now = cfs_time_current_sec();
 983
 984                 if (cfs_time_after(now, ratelimit)) {
 985                         CWARN("upcall timed out, is lgssd running?\n");
 986                         ratelimit = now + 15;
 987                 }
 988         }
 989         gss_msg_fail_ctx(gmsg);
 990         gss_release_msg(gmsg);
 991         EXIT;
 992 }
 993
 994 static
 995 void gss_pipe_release(struct inode *inode)
 996 {
 997         struct rpc_inode *rpci = RPC_I(inode);
 998         __u32             idx;
 999         ENTRY;
1000
1001         idx = (__u32) (long) rpci->private;
1002         LASSERT(idx < MECH_MAX);
1003
1004         upcall_list_lock(idx);
1005         while (!cfs_list_empty(&upcall_lists[idx])) {
1006                 struct gss_upcall_msg      *gmsg;
1007                 struct gss_upcall_msg_data *gumd;
1008
1009                 gmsg = cfs_list_entry(upcall_lists[idx].next,
1010                                       struct gss_upcall_msg, gum_list);
1011                 gumd = &gmsg->gum_data;
1012                 LASSERT(cfs_list_empty(&gmsg->gum_base.list));
1013
1014                 CERROR("failing remaining msg %p:seq %u, uid %u, svc %u, "
1015                        "nid "LPX64", obd %.*s\n", gmsg,
1016                        gumd->gum_seq, gumd->gum_uid, gumd->gum_svc,
1017                        gumd->gum_nid, (int) sizeof(gumd->gum_obd),
1018                        gumd->gum_obd);
1019
1020                 gmsg->gum_base.errno = -EPIPE;
1021                 cfs_atomic_inc(&gmsg->gum_refcount);
1022                 gss_unhash_msg_nolock(gmsg);
1023
1024                 gss_msg_fail_ctx(gmsg);
1025
1026                 upcall_list_unlock(idx);
1027                 gss_release_msg(gmsg);
1028                 upcall_list_lock(idx);
1029         }
1030         upcall_list_unlock(idx);
1031         EXIT;
1032 }
1033
1034 static struct rpc_pipe_ops gss_upcall_ops = {
1035         .upcall         = gss_pipe_upcall,
1036         .downcall       = gss_pipe_downcall,
1037         .destroy_msg    = gss_pipe_destroy_msg,
1038         .release_pipe   = gss_pipe_release,
1039 };
1040
1041 /****************************************
1042  * upcall helper functions              *
1043  ****************************************/
1044
1045 static
1046 int gss_ctx_refresh_pf(struct ptlrpc_cli_ctx *ctx)
1047 {
1048         struct obd_import          *imp;
1049         struct gss_sec             *gsec;
1050         struct gss_upcall_msg      *gmsg;
1051         int                         rc = 0;
1052         ENTRY;
1053
1054         cfs_might_sleep();
1055
1056         LASSERT(ctx->cc_sec);
1057         LASSERT(ctx->cc_sec->ps_import);
1058         LASSERT(ctx->cc_sec->ps_import->imp_obd);
1059
1060         imp = ctx->cc_sec->ps_import;
1061         if (!imp->imp_connection) {
1062                 CERROR("import has no connection set\n");
1063                 RETURN(-EINVAL);
1064         }
1065
1066         gsec = container_of(ctx->cc_sec, struct gss_sec, gs_base);
1067
1068         OBD_ALLOC_PTR(gmsg);
1069         if (!gmsg)
1070                 RETURN(-ENOMEM);
1071
1072         /* initialize pipefs base msg */
1073         CFS_INIT_LIST_HEAD(&gmsg->gum_base.list);
1074         gmsg->gum_base.data = &gmsg->gum_data;
1075         gmsg->gum_base.len = sizeof(gmsg->gum_data);
1076         gmsg->gum_base.copied = 0;
1077         gmsg->gum_base.errno = 0;
1078
1079         /* init upcall msg */
1080         cfs_atomic_set(&gmsg->gum_refcount, 1);
1081         gmsg->gum_mechidx = mech_name2idx(gsec->gs_mech->gm_name);
1082         gmsg->gum_gsec = gsec;
1083         gmsg->gum_gctx = container_of(sptlrpc_cli_ctx_get(ctx),
1084                                       struct gss_cli_ctx, gc_base);
1085         gmsg->gum_data.gum_seq = upcall_get_sequence();
1086         gmsg->gum_data.gum_uid = ctx->cc_vcred.vc_uid;
1087         gmsg->gum_data.gum_gid = 0; /* not used for now */
1088         gmsg->gum_data.gum_svc = import_to_gss_svc(imp);
1089         gmsg->gum_data.gum_nid = imp->imp_connection->c_peer.nid;
1090         strncpy(gmsg->gum_data.gum_obd, imp->imp_obd->obd_name,
1091                 sizeof(gmsg->gum_data.gum_obd));
1092
1093         /* This only could happen when sysadmin set it dead/expired
1094          * using lctl by force. */
1095         if (ctx->cc_flags & PTLRPC_CTX_STATUS_MASK) {
1096                 CWARN("ctx %p(%u->%s) was set flags %lx unexpectedly\n",
1097                       ctx, ctx->cc_vcred.vc_uid, sec2target_str(ctx->cc_sec),
1098                       ctx->cc_flags);
1099
1100                 LASSERT(!(ctx->cc_flags & PTLRPC_CTX_UPTODATE));
1101                 ctx->cc_flags |= PTLRPC_CTX_DEAD | PTLRPC_CTX_ERROR;
1102
1103                 rc = -EIO;
1104                 goto err_free;
1105         }
1106
1107         upcall_msg_enlist(gmsg);
1108
1109         rc = rpc_queue_upcall(de_pipes[gmsg->gum_mechidx]->d_inode,
1110                               &gmsg->gum_base);
1111         if (rc) {
1112                 CERROR("rpc_queue_upcall failed: %d\n", rc);
1113
1114                 upcall_msg_delist(gmsg);
1115                 goto err_free;
1116         }
1117
1118         RETURN(0);
1119 err_free:
1120         OBD_FREE_PTR(gmsg);
1121         RETURN(rc);
1122 }
1123
1124 static
1125 int gss_cli_ctx_refresh_pf(struct ptlrpc_cli_ctx *ctx)
1126 {
1127         /* if we are refreshing for root, also update the reverse
1128          * handle index, do not confuse reverse contexts. */
1129         if (ctx->cc_vcred.vc_uid == 0) {
1130                 struct gss_sec *gsec;
1131
1132                 gsec = container_of(ctx->cc_sec, struct gss_sec, gs_base);
1133                 gsec->gs_rvs_hdl = gss_get_next_ctx_index();
1134         }
1135
1136         return gss_ctx_refresh_pf(ctx);
1137 }
1138
1139 /****************************************
1140  * lustre gss pipefs policy             *
1141  ****************************************/
1142
1143 static struct ptlrpc_ctx_ops gss_pipefs_ctxops = {
1144         .match                  = gss_cli_ctx_match,
1145         .refresh                = gss_cli_ctx_refresh_pf,
1146         .validate               = gss_cli_ctx_validate_pf,
1147         .die                    = gss_cli_ctx_die_pf,
1148         .sign                   = gss_cli_ctx_sign,
1149         .verify                 = gss_cli_ctx_verify,
1150         .seal                   = gss_cli_ctx_seal,
1151         .unseal                 = gss_cli_ctx_unseal,
1152         .wrap_bulk              = gss_cli_ctx_wrap_bulk,
1153         .unwrap_bulk            = gss_cli_ctx_unwrap_bulk,
1154 };
1155
1156 static struct ptlrpc_sec_cops gss_sec_pipefs_cops = {
1157         .create_sec             = gss_sec_create_pf,
1158         .destroy_sec            = gss_sec_destroy_pf,
1159         .kill_sec               = gss_sec_kill,
1160         .lookup_ctx             = gss_sec_lookup_ctx_pf,
1161         .release_ctx            = gss_sec_release_ctx_pf,
1162         .flush_ctx_cache        = gss_sec_flush_ctx_cache_pf,
1163         .install_rctx           = gss_sec_install_rctx,
1164         .alloc_reqbuf           = gss_alloc_reqbuf,
1165         .free_reqbuf            = gss_free_reqbuf,
1166         .alloc_repbuf           = gss_alloc_repbuf,
1167         .free_repbuf            = gss_free_repbuf,
1168         .enlarge_reqbuf         = gss_enlarge_reqbuf,
1169 };
1170
1171 static struct ptlrpc_sec_sops gss_sec_pipefs_sops = {
1172         .accept                 = gss_svc_accept_pf,
1173         .invalidate_ctx         = gss_svc_invalidate_ctx,
1174         .alloc_rs               = gss_svc_alloc_rs,
1175         .authorize              = gss_svc_authorize,
1176         .free_rs                = gss_svc_free_rs,
1177         .free_ctx               = gss_svc_free_ctx,
1178         .unwrap_bulk            = gss_svc_unwrap_bulk,
1179         .wrap_bulk              = gss_svc_wrap_bulk,
1180         .install_rctx           = gss_svc_install_rctx_pf,
1181 };
1182
1183 static struct ptlrpc_sec_policy gss_policy_pipefs = {
1184         .sp_owner               = THIS_MODULE,
1185         .sp_name                = "gss.pipefs",
1186         .sp_policy              = SPTLRPC_POLICY_GSS_PIPEFS,
1187         .sp_cops                = &gss_sec_pipefs_cops,
1188         .sp_sops                = &gss_sec_pipefs_sops,
1189 };
1190
1191 static
1192 int __init gss_init_pipefs_upcall(void)
1193 {
1194         struct dentry   *de;
1195
1196         /* pipe dir */
1197         de = rpc_mkdir(LUSTRE_PIPE_ROOT, NULL);
1198         if (IS_ERR(de) && PTR_ERR(de) != -EEXIST) {
1199                 CERROR("Failed to create gss pipe dir: %ld\n", PTR_ERR(de));
1200                 return PTR_ERR(de);
1201         }
1202
1203         /* FIXME hack pipefs: dput will sometimes cause oops during module
1204          * unload and lgssd close the pipe fds. */
1205
1206         /* krb5 mechanism */
1207         de = rpc_mkpipe(LUSTRE_PIPE_KRB5, (void *) MECH_KRB5, &gss_upcall_ops,
1208                         RPC_PIPE_WAIT_FOR_OPEN);
1209         if (!de || IS_ERR(de)) {
1210                 CERROR("failed to make rpc_pipe %s: %ld\n",
1211                        LUSTRE_PIPE_KRB5, PTR_ERR(de));
1212                 rpc_rmdir(LUSTRE_PIPE_ROOT);
1213                 return PTR_ERR(de);
1214         }
1215
1216         de_pipes[MECH_KRB5] = de;
1217         CFS_INIT_LIST_HEAD(&upcall_lists[MECH_KRB5]);
1218         cfs_spin_lock_init(&upcall_locks[MECH_KRB5]);
1219
1220         return 0;
1221 }
1222
1223 static
1224 void __exit gss_exit_pipefs_upcall(void)
1225 {
1226         __u32   i;
1227
1228         for (i = 0; i < MECH_MAX; i++) {
1229                 LASSERT(cfs_list_empty(&upcall_lists[i]));
1230
1231                 /* dput pipe dentry here might cause lgssd oops. */
1232                 de_pipes[i] = NULL;
1233         }
1234
1235         rpc_unlink(LUSTRE_PIPE_KRB5);
1236         rpc_rmdir(LUSTRE_PIPE_ROOT);
1237 }
1238
1239 int __init gss_init_pipefs(void)
1240 {
1241         int rc;
1242
1243         rc = gss_init_pipefs_upcall();
1244         if (rc)
1245                 return rc;
1246
1247         rc = sptlrpc_register_policy(&gss_policy_pipefs);
1248         if (rc) {
1249                 gss_exit_pipefs_upcall();
1250                 return rc;
1251         }
1252
1253         return 0;
1254 }
1255
1256 void __exit gss_exit_pipefs(void)
1257 {
1258         gss_exit_pipefs_upcall();
1259         sptlrpc_unregister_policy(&gss_policy_pipefs);
1260 }