lustre/ptlrpc/gss/gss_pipefs.c

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