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