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