lustre/ptlrpc/gss/gss_svc_upcall.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.
   7  * Author: Eric Mei <eric.mei@sun.com>
   8  *
   9  * Copyright 2004 - 2006, Cluster File Systems, Inc.
  10  * All rights reserved
  11  * Author: Eric Mei <ericm@clusterfs.com>
  12  */
  13
  14 /*
  15  * Neil Brown <neilb@cse.unsw.edu.au>
  16  * J. Bruce Fields <bfields@umich.edu>
  17  * Andy Adamson <andros@umich.edu>
  18  * Dug Song <dugsong@monkey.org>
  19  *
  20  * RPCSEC_GSS server authentication.
  21  * This implements RPCSEC_GSS as defined in rfc2203 (rpcsec_gss) and rfc2078
  22  * (gssapi)
  23  *
  24  * The RPCSEC_GSS involves three stages:
  25  *  1/ context creation
  26  *  2/ data exchange
  27  *  3/ context destruction
  28  *
  29  * Context creation is handled largely by upcalls to user-space.
  30  *  In particular, GSS_Accept_sec_context is handled by an upcall
  31  * Data exchange is handled entirely within the kernel
  32  *  In particular, GSS_GetMIC, GSS_VerifyMIC, GSS_Seal, GSS_Unseal are in-kernel.
  33  * Context destruction is handled in-kernel
  34  *  GSS_Delete_sec_context is in-kernel
  35  *
  36  * Context creation is initiated by a RPCSEC_GSS_INIT request arriving.
  37  * The context handle and gss_token are used as a key into the rpcsec_init cache.
  38  * The content of this cache includes some of the outputs of GSS_Accept_sec_context,
  39  * being major_status, minor_status, context_handle, reply_token.
  40  * These are sent back to the client.
  41  * Sequence window management is handled by the kernel.  The window size if currently
  42  * a compile time constant.
  43  *
  44  * When user-space is happy that a context is established, it places an entry
  45  * in the rpcsec_context cache. The key for this cache is the context_handle.
  46  * The content includes:
  47  *   uid/gidlist - for determining access rights
  48  *   mechanism type
  49  *   mechanism specific information, such as a key
  50  *
  51  */
  52
  53 #define DEBUG_SUBSYSTEM S_SEC
  54 #ifdef __KERNEL__
  55 #include <linux/types.h>
  56 #include <linux/init.h>
  57 #include <linux/module.h>
  58 #include <linux/slab.h>
  59 #include <linux/hash.h>
  60 #include <linux/mutex.h>
  61 #include <linux/sunrpc/cache.h>
  62 #else
  63 #include <liblustre.h>
  64 #endif
  65
  66 #include <obd.h>
  67 #include <obd_class.h>
  68 #include <obd_support.h>
  69 #include <lustre/lustre_idl.h>
  70 #include <lustre_net.h>
  71 #include <lustre_import.h>
  72 #include <lustre_sec.h>
  73
  74 #include "gss_err.h"
  75 #include "gss_internal.h"
  76 #include "gss_api.h"
  77
  78 #define GSS_SVC_UPCALL_TIMEOUT  (20)
  79
  80 static spinlock_t __ctx_index_lock = SPIN_LOCK_UNLOCKED;
  81 static __u64 __ctx_index;
  82
  83 __u64 gss_get_next_ctx_index(void)
  84 {
  85         __u64 idx;
  86
  87         spin_lock(&__ctx_index_lock);
  88         idx = __ctx_index++;
  89         spin_unlock(&__ctx_index_lock);
  90
  91         return idx;
  92 }
  93
  94 static inline unsigned long hash_mem(char *buf, int length, int bits)
  95 {
  96         unsigned long hash = 0;
  97         unsigned long l = 0;
  98         int len = 0;
  99         unsigned char c;
 100
 101         do {
 102                 if (len == length) {
 103                         c = (char) len;
 104                         len = -1;
 105                 } else
 106                         c = *buf++;
 107
 108                 l = (l << 8) | c;
 109                 len++;
 110
 111                 if ((len & (BITS_PER_LONG/8-1)) == 0)
 112                         hash = hash_long(hash^l, BITS_PER_LONG);
 113         } while (len);
 114
 115         return hash >> (BITS_PER_LONG - bits);
 116 }
 117
 118 /****************************************
 119  * rsi cache                            *
 120  ****************************************/
 121
 122 #define RSI_HASHBITS    (6)
 123 #define RSI_HASHMAX     (1 << RSI_HASHBITS)
 124 #define RSI_HASHMASK    (RSI_HASHMAX - 1)
 125
 126 struct rsi {
 127         struct cache_head       h;
 128         __u32                   lustre_svc;
 129         __u64                   nid;
 130         wait_queue_head_t       waitq;
 131         rawobj_t                in_handle, in_token;
 132         rawobj_t                out_handle, out_token;
 133         int                     major_status, minor_status;
 134 };
 135
 136 static struct cache_head *rsi_table[RSI_HASHMAX];
 137 static struct cache_detail rsi_cache;
 138 #ifdef HAVE_SUNRPC_CACHE_V2
 139 static struct rsi *rsi_update(struct rsi *new, struct rsi *old);
 140 static struct rsi *rsi_lookup(struct rsi *item);
 141 #else
 142 static struct rsi *rsi_lookup(struct rsi *item, int set);
 143 #endif
 144
 145 static inline int rsi_hash(struct rsi *item)
 146 {
 147         return hash_mem((char *)item->in_handle.data, item->in_handle.len,
 148                         RSI_HASHBITS) ^
 149                hash_mem((char *)item->in_token.data, item->in_token.len,
 150                         RSI_HASHBITS);
 151 }
 152
 153 static inline int __rsi_match(struct rsi *item, struct rsi *tmp)
 154 {
 155         return (rawobj_equal(&item->in_handle, &tmp->in_handle) &&
 156                 rawobj_equal(&item->in_token, &tmp->in_token));
 157 }
 158
 159 static void rsi_free(struct rsi *rsi)
 160 {
 161         rawobj_free(&rsi->in_handle);
 162         rawobj_free(&rsi->in_token);
 163         rawobj_free(&rsi->out_handle);
 164         rawobj_free(&rsi->out_token);
 165 }
 166
 167 static void rsi_request(struct cache_detail *cd,
 168                         struct cache_head *h,
 169                         char **bpp, int *blen)
 170 {
 171         struct rsi *rsi = container_of(h, struct rsi, h);
 172         __u64 index = 0;
 173
 174         /* if in_handle is null, provide kernel suggestion */
 175         if (rsi->in_handle.len == 0)
 176                 index = gss_get_next_ctx_index();
 177
 178         qword_addhex(bpp, blen, (char *) &rsi->lustre_svc,
 179                      sizeof(rsi->lustre_svc));
 180         qword_addhex(bpp, blen, (char *) &rsi->nid, sizeof(rsi->nid));
 181         qword_addhex(bpp, blen, (char *) &index, sizeof(index));
 182         qword_addhex(bpp, blen, rsi->in_handle.data, rsi->in_handle.len);
 183         qword_addhex(bpp, blen, rsi->in_token.data, rsi->in_token.len);
 184         (*bpp)[-1] = '\n';
 185 }
 186
 187 static inline void __rsi_init(struct rsi *new, struct rsi *item)
 188 {
 189         new->out_handle = RAWOBJ_EMPTY;
 190         new->out_token = RAWOBJ_EMPTY;
 191
 192         new->in_handle = item->in_handle;
 193         item->in_handle = RAWOBJ_EMPTY;
 194         new->in_token = item->in_token;
 195         item->in_token = RAWOBJ_EMPTY;
 196
 197         new->lustre_svc = item->lustre_svc;
 198         new->nid = item->nid;
 199         init_waitqueue_head(&new->waitq);
 200 }
 201
 202 static inline void __rsi_update(struct rsi *new, struct rsi *item)
 203 {
 204         LASSERT(new->out_handle.len == 0);
 205         LASSERT(new->out_token.len == 0);
 206
 207         new->out_handle = item->out_handle;
 208         item->out_handle = RAWOBJ_EMPTY;
 209         new->out_token = item->out_token;
 210         item->out_token = RAWOBJ_EMPTY;
 211
 212         new->major_status = item->major_status;
 213         new->minor_status = item->minor_status;
 214 }
 215
 216 #ifdef HAVE_SUNRPC_CACHE_V2
 217
 218 static void rsi_put(struct kref *ref)
 219 {
 220         struct rsi *rsi = container_of(ref, struct rsi, h.ref);
 221
 222         LASSERT(rsi->h.next == NULL);
 223         rsi_free(rsi);
 224         OBD_FREE_PTR(rsi);
 225 }
 226
 227 static int rsi_match(struct cache_head *a, struct cache_head *b)
 228 {
 229         struct rsi *item = container_of(a, struct rsi, h);
 230         struct rsi *tmp = container_of(b, struct rsi, h);
 231
 232         return __rsi_match(item, tmp);
 233 }
 234
 235 static void rsi_init(struct cache_head *cnew, struct cache_head *citem)
 236 {
 237         struct rsi *new = container_of(cnew, struct rsi, h);
 238         struct rsi *item = container_of(citem, struct rsi, h);
 239
 240         __rsi_init(new, item);
 241 }
 242
 243 static void update_rsi(struct cache_head *cnew, struct cache_head *citem)
 244 {
 245         struct rsi *new = container_of(cnew, struct rsi, h);
 246         struct rsi *item = container_of(citem, struct rsi, h);
 247
 248         __rsi_update(new, item);
 249 }
 250
 251 static struct cache_head *rsi_alloc(void)
 252 {
 253         struct rsi *rsi;
 254
 255         OBD_ALLOC_PTR(rsi);
 256         if (rsi)
 257                 return &rsi->h;
 258         else
 259                 return NULL;
 260 }
 261
 262 static int rsi_parse(struct cache_detail *cd, char *mesg, int mlen)
 263 {
 264         char           *buf = mesg;
 265         char           *ep;
 266         int             len;
 267         struct rsi      rsii, *rsip = NULL;
 268         time_t          expiry;
 269         int             status = -EINVAL;
 270         ENTRY;
 271
 272
 273         memset(&rsii, 0, sizeof(rsii));
 274
 275         /* handle */
 276         len = qword_get(&mesg, buf, mlen);
 277         if (len < 0)
 278                 goto out;
 279         if (rawobj_alloc(&rsii.in_handle, buf, len)) {
 280                 status = -ENOMEM;
 281                 goto out;
 282         }
 283
 284         /* token */
 285         len = qword_get(&mesg, buf, mlen);
 286         if (len < 0)
 287                 goto out;
 288         if (rawobj_alloc(&rsii.in_token, buf, len)) {
 289                 status = -ENOMEM;
 290                 goto out;
 291         }
 292
 293         rsip = rsi_lookup(&rsii);
 294         if (!rsip)
 295                 goto out;
 296
 297         rsii.h.flags = 0;
 298         /* expiry */
 299         expiry = get_expiry(&mesg);
 300         if (expiry == 0)
 301                 goto out;
 302
 303         len = qword_get(&mesg, buf, mlen);
 304         if (len <= 0)
 305                 goto out;
 306
 307         /* major */
 308         rsii.major_status = simple_strtol(buf, &ep, 10);
 309         if (*ep)
 310                 goto out;
 311
 312         /* minor */
 313         len = qword_get(&mesg, buf, mlen);
 314         if (len <= 0)
 315                 goto out;
 316         rsii.minor_status = simple_strtol(buf, &ep, 10);
 317         if (*ep)
 318                 goto out;
 319
 320         /* out_handle */
 321         len = qword_get(&mesg, buf, mlen);
 322         if (len < 0)
 323                 goto out;
 324         if (rawobj_alloc(&rsii.out_handle, buf, len)) {
 325                 status = -ENOMEM;
 326                 goto out;
 327         }
 328
 329         /* out_token */
 330         len = qword_get(&mesg, buf, mlen);
 331         if (len < 0)
 332                 goto out;
 333         if (rawobj_alloc(&rsii.out_token, buf, len)) {
 334                 status = -ENOMEM;
 335                 goto out;
 336         }
 337
 338         rsii.h.expiry_time = expiry;
 339         rsip = rsi_update(&rsii, rsip);
 340         status = 0;
 341 out:
 342         rsi_free(&rsii);
 343         if (rsip) {
 344                 wake_up_all(&rsip->waitq);
 345                 cache_put(&rsip->h, &rsi_cache);
 346         } else {
 347                 status = -ENOMEM;
 348         }
 349
 350         if (status)
 351                 CERROR("rsi parse error %d\n", status);
 352         RETURN(status);
 353 }
 354
 355 #else /* !HAVE_SUNRPC_CACHE_V2 */
 356
 357 static void rsi_put(struct cache_head *item, struct cache_detail *cd)
 358 {
 359         struct rsi *rsi = container_of(item, struct rsi, h);
 360
 361         LASSERT(atomic_read(&item->refcnt) > 0);
 362
 363         if (cache_put(item, cd)) {
 364                 LASSERT(item->next == NULL);
 365                 rsi_free(rsi);
 366                 kfree(rsi); /* created by cache mgmt using kmalloc */
 367         }
 368 }
 369
 370 static inline int rsi_match(struct rsi *item, struct rsi *tmp)
 371 {
 372         return __rsi_match(item, tmp);
 373 }
 374
 375 static inline void rsi_init(struct rsi *new, struct rsi *item)
 376 {
 377         __rsi_init(new, item);
 378 }
 379
 380 static inline void rsi_update(struct rsi *new, struct rsi *item)
 381 {
 382         __rsi_update(new, item);
 383 }
 384
 385 static int rsi_parse(struct cache_detail *cd, char *mesg, int mlen)
 386 {
 387         char           *buf = mesg;
 388         char           *ep;
 389         int             len;
 390         struct rsi      rsii, *rsip = NULL;
 391         time_t          expiry;
 392         int             status = -EINVAL;
 393         ENTRY;
 394
 395
 396         memset(&rsii, 0, sizeof(rsii));
 397
 398         /* handle */
 399         len = qword_get(&mesg, buf, mlen);
 400         if (len < 0)
 401                 goto out;
 402         if (rawobj_alloc(&rsii.in_handle, buf, len)) {
 403                 status = -ENOMEM;
 404                 goto out;
 405         }
 406
 407         /* token */
 408         len = qword_get(&mesg, buf, mlen);
 409         if (len < 0)
 410                 goto out;
 411         if (rawobj_alloc(&rsii.in_token, buf, len)) {
 412                 status = -ENOMEM;
 413                 goto out;
 414         }
 415
 416         /* expiry */
 417         expiry = get_expiry(&mesg);
 418         if (expiry == 0)
 419                 goto out;
 420
 421         len = qword_get(&mesg, buf, mlen);
 422         if (len <= 0)
 423                 goto out;
 424
 425         /* major */
 426         rsii.major_status = simple_strtol(buf, &ep, 10);
 427         if (*ep)
 428                 goto out;
 429
 430         /* minor */
 431         len = qword_get(&mesg, buf, mlen);
 432         if (len <= 0)
 433                 goto out;
 434         rsii.minor_status = simple_strtol(buf, &ep, 10);
 435         if (*ep)
 436                 goto out;
 437
 438         /* out_handle */
 439         len = qword_get(&mesg, buf, mlen);
 440         if (len < 0)
 441                 goto out;
 442         if (rawobj_alloc(&rsii.out_handle, buf, len)) {
 443                 status = -ENOMEM;
 444                 goto out;
 445         }
 446
 447         /* out_token */
 448         len = qword_get(&mesg, buf, mlen);
 449         if (len < 0)
 450                 goto out;
 451         if (rawobj_alloc(&rsii.out_token, buf, len)) {
 452                 status = -ENOMEM;
 453                 goto out;
 454         }
 455
 456         rsii.h.expiry_time = expiry;
 457         rsip = rsi_lookup(&rsii, 1);
 458         status = 0;
 459 out:
 460         rsi_free(&rsii);
 461         if (rsip) {
 462                 wake_up_all(&rsip->waitq);
 463                 rsi_put(&rsip->h, &rsi_cache);
 464         }
 465
 466         if (status)
 467                 CERROR("rsi parse error %d\n", status);
 468         RETURN(status);
 469 }
 470
 471 #endif /* HAVE_SUNRPC_CACHE_V2 */
 472
 473 static struct cache_detail rsi_cache = {
 474         .hash_size      = RSI_HASHMAX,
 475         .hash_table     = rsi_table,
 476         .name           = "auth.sptlrpc.init",
 477         .cache_put      = rsi_put,
 478         .cache_request  = rsi_request,
 479         .cache_parse    = rsi_parse,
 480 #ifdef HAVE_SUNRPC_CACHE_V2
 481         .match          = rsi_match,
 482         .init           = rsi_init,
 483         .update         = update_rsi,
 484         .alloc          = rsi_alloc,
 485 #endif
 486 };
 487
 488 #ifdef HAVE_SUNRPC_CACHE_V2
 489
 490 static struct rsi *rsi_lookup(struct rsi *item)
 491 {
 492         struct cache_head *ch;
 493         int hash = rsi_hash(item);
 494
 495         ch = sunrpc_cache_lookup(&rsi_cache, &item->h, hash);
 496         if (ch)
 497                 return container_of(ch, struct rsi, h);
 498         else
 499                 return NULL;
 500 }
 501
 502 static struct rsi *rsi_update(struct rsi *new, struct rsi *old)
 503 {
 504         struct cache_head *ch;
 505         int hash = rsi_hash(new);
 506
 507         ch = sunrpc_cache_update(&rsi_cache, &new->h, &old->h, hash);
 508         if (ch)
 509                 return container_of(ch, struct rsi, h);
 510         else
 511                 return NULL;
 512 }
 513
 514 #else
 515
 516 static DefineSimpleCacheLookup(rsi, 0)
 517
 518 #endif
 519
 520 /****************************************
 521  * rsc cache                            *
 522  ****************************************/
 523
 524 #define RSC_HASHBITS    (10)
 525 #define RSC_HASHMAX     (1 << RSC_HASHBITS)
 526 #define RSC_HASHMASK    (RSC_HASHMAX - 1)
 527
 528 struct rsc {
 529         struct cache_head       h;
 530         struct obd_device      *target;
 531         rawobj_t                handle;
 532         struct gss_svc_ctx      ctx;
 533 };
 534
 535 static struct cache_head *rsc_table[RSC_HASHMAX];
 536 static struct cache_detail rsc_cache;
 537 #ifdef HAVE_SUNRPC_CACHE_V2
 538 static struct rsc *rsc_update(struct rsc *new, struct rsc *old);
 539 static struct rsc *rsc_lookup(struct rsc *item);
 540 #else
 541 static struct rsc *rsc_lookup(struct rsc *item, int set);
 542 #endif
 543
 544 static void rsc_free(struct rsc *rsci)
 545 {
 546         rawobj_free(&rsci->handle);
 547         rawobj_free(&rsci->ctx.gsc_rvs_hdl);
 548         lgss_delete_sec_context(&rsci->ctx.gsc_mechctx);
 549 }
 550
 551 static inline int rsc_hash(struct rsc *rsci)
 552 {
 553         return hash_mem((char *)rsci->handle.data,
 554                         rsci->handle.len, RSC_HASHBITS);
 555 }
 556
 557 static inline int __rsc_match(struct rsc *new, struct rsc *tmp)
 558 {
 559         return rawobj_equal(&new->handle, &tmp->handle);
 560 }
 561
 562 static inline void __rsc_init(struct rsc *new, struct rsc *tmp)
 563 {
 564         new->handle = tmp->handle;
 565         tmp->handle = RAWOBJ_EMPTY;
 566
 567         new->target = NULL;
 568         memset(&new->ctx, 0, sizeof(new->ctx));
 569         new->ctx.gsc_rvs_hdl = RAWOBJ_EMPTY;
 570 }
 571
 572 static inline void __rsc_update(struct rsc *new, struct rsc *tmp)
 573 {
 574         new->ctx = tmp->ctx;
 575         tmp->ctx.gsc_rvs_hdl = RAWOBJ_EMPTY;
 576         tmp->ctx.gsc_mechctx = NULL;
 577
 578         memset(&new->ctx.gsc_seqdata, 0, sizeof(new->ctx.gsc_seqdata));
 579         spin_lock_init(&new->ctx.gsc_seqdata.ssd_lock);
 580 }
 581
 582 #ifdef HAVE_SUNRPC_CACHE_V2
 583
 584 static void rsc_put(struct kref *ref)
 585 {
 586         struct rsc *rsci = container_of(ref, struct rsc, h.ref);
 587
 588         LASSERT(rsci->h.next == NULL);
 589         rsc_free(rsci);
 590         OBD_FREE_PTR(rsci);
 591 }
 592
 593 static int rsc_match(struct cache_head *a, struct cache_head *b)
 594 {
 595         struct rsc *new = container_of(a, struct rsc, h);
 596         struct rsc *tmp = container_of(b, struct rsc, h);
 597
 598         return __rsc_match(new, tmp);
 599 }
 600
 601 static void rsc_init(struct cache_head *cnew, struct cache_head *ctmp)
 602 {
 603         struct rsc *new = container_of(cnew, struct rsc, h);
 604         struct rsc *tmp = container_of(ctmp, struct rsc, h);
 605
 606         __rsc_init(new, tmp);
 607 }
 608
 609 static void update_rsc(struct cache_head *cnew, struct cache_head *ctmp)
 610 {
 611         struct rsc *new = container_of(cnew, struct rsc, h);
 612         struct rsc *tmp = container_of(ctmp, struct rsc, h);
 613
 614         __rsc_update(new, tmp);
 615 }
 616
 617 static struct cache_head * rsc_alloc(void)
 618 {
 619         struct rsc *rsc;
 620
 621         OBD_ALLOC_PTR(rsc);
 622         if (rsc)
 623                 return &rsc->h;
 624         else
 625                 return NULL;
 626 }
 627
 628 static int rsc_parse(struct cache_detail *cd, char *mesg, int mlen)
 629 {
 630         char                *buf = mesg;
 631         int                  len, rv, tmp_int;
 632         struct rsc           rsci, *rscp = NULL;
 633         time_t               expiry;
 634         int                  status = -EINVAL;
 635         struct gss_api_mech *gm = NULL;
 636
 637         memset(&rsci, 0, sizeof(rsci));
 638
 639         /* context handle */
 640         len = qword_get(&mesg, buf, mlen);
 641         if (len < 0) goto out;
 642         status = -ENOMEM;
 643         if (rawobj_alloc(&rsci.handle, buf, len))
 644                 goto out;
 645
 646         rsci.h.flags = 0;
 647         /* expiry */
 648         expiry = get_expiry(&mesg);
 649         status = -EINVAL;
 650         if (expiry == 0)
 651                 goto out;
 652
 653         /* remote flag */
 654         rv = get_int(&mesg, &tmp_int);
 655         if (rv) {
 656                 CERROR("fail to get remote flag\n");
 657                 goto out;
 658         }
 659         rsci.ctx.gsc_remote = (tmp_int != 0);
 660
 661         /* root user flag */
 662         rv = get_int(&mesg, &tmp_int);
 663         if (rv) {
 664                 CERROR("fail to get oss user flag\n");
 665                 goto out;
 666         }
 667         rsci.ctx.gsc_usr_root = (tmp_int != 0);
 668
 669         /* mds user flag */
 670         rv = get_int(&mesg, &tmp_int);
 671         if (rv) {
 672                 CERROR("fail to get mds user flag\n");
 673                 goto out;
 674         }
 675         rsci.ctx.gsc_usr_mds = (tmp_int != 0);
 676
 677         /* mapped uid */
 678         rv = get_int(&mesg, (int *) &rsci.ctx.gsc_mapped_uid);
 679         if (rv) {
 680                 CERROR("fail to get mapped uid\n");
 681                 goto out;
 682         }
 683
 684         rscp = rsc_lookup(&rsci);
 685         if (!rscp)
 686                 goto out;
 687
 688         /* uid, or NEGATIVE */
 689         rv = get_int(&mesg, (int *) &rsci.ctx.gsc_uid);
 690         if (rv == -EINVAL)
 691                 goto out;
 692         if (rv == -ENOENT) {
 693                 CERROR("NOENT? set rsc entry negative\n");
 694                 set_bit(CACHE_NEGATIVE, &rsci.h.flags);
 695         } else {
 696                 rawobj_t tmp_buf;
 697                 unsigned long ctx_expiry;
 698
 699                 /* gid */
 700                 if (get_int(&mesg, (int *) &rsci.ctx.gsc_gid))
 701                         goto out;
 702
 703                 /* mech name */
 704                 len = qword_get(&mesg, buf, mlen);
 705                 if (len < 0)
 706                         goto out;
 707                 gm = lgss_name_to_mech(buf);
 708                 status = -EOPNOTSUPP;
 709                 if (!gm)
 710                         goto out;
 711
 712                 status = -EINVAL;
 713                 /* mech-specific data: */
 714                 len = qword_get(&mesg, buf, mlen);
 715                 if (len < 0)
 716                         goto out;
 717
 718                 tmp_buf.len = len;
 719                 tmp_buf.data = (unsigned char *)buf;
 720                 if (lgss_import_sec_context(&tmp_buf, gm,
 721                                             &rsci.ctx.gsc_mechctx))
 722                         goto out;
 723
 724                 /* currently the expiry time passed down from user-space
 725                  * is invalid, here we retrive it from mech. */
 726                 if (lgss_inquire_context(rsci.ctx.gsc_mechctx, &ctx_expiry)) {
 727                         CERROR("unable to get expire time, drop it\n");
 728                         goto out;
 729                 }
 730                 expiry = (time_t) ctx_expiry;
 731         }
 732
 733         rsci.h.expiry_time = expiry;
 734         rscp = rsc_update(&rsci, rscp);
 735         status = 0;
 736 out:
 737         if (gm)
 738                 lgss_mech_put(gm);
 739         rsc_free(&rsci);
 740         if (rscp)
 741                 cache_put(&rscp->h, &rsc_cache);
 742         else
 743                 status = -ENOMEM;
 744
 745         if (status)
 746                 CERROR("parse rsc error %d\n", status);
 747         return status;
 748 }
 749
 750 #else /* !HAVE_SUNRPC_CACHE_V2 */
 751
 752 static void rsc_put(struct cache_head *item, struct cache_detail *cd)
 753 {
 754         struct rsc *rsci = container_of(item, struct rsc, h);
 755
 756         LASSERT(atomic_read(&item->refcnt) > 0);
 757
 758         if (cache_put(item, cd)) {
 759                 LASSERT(item->next == NULL);
 760                 rsc_free(rsci);
 761                 kfree(rsci); /* created by cache mgmt using kmalloc */
 762         }
 763 }
 764
 765 static inline int rsc_match(struct rsc *new, struct rsc *tmp)
 766 {
 767         return __rsc_match(new, tmp);
 768 }
 769
 770 static inline void rsc_init(struct rsc *new, struct rsc *tmp)
 771 {
 772         __rsc_init(new, tmp);
 773 }
 774
 775 static inline void rsc_update(struct rsc *new, struct rsc *tmp)
 776 {
 777         __rsc_update(new, tmp);
 778 }
 779
 780 static int rsc_parse(struct cache_detail *cd, char *mesg, int mlen)
 781 {
 782         char       *buf = mesg;
 783         int         len, rv, tmp_int;
 784         struct rsc  rsci, *rscp = NULL;
 785         time_t      expiry;
 786         int         status = -EINVAL;
 787
 788         memset(&rsci, 0, sizeof(rsci));
 789
 790         /* context handle */
 791         len = qword_get(&mesg, buf, mlen);
 792         if (len < 0) goto out;
 793         status = -ENOMEM;
 794         if (rawobj_alloc(&rsci.handle, buf, len))
 795                 goto out;
 796
 797         rsci.h.flags = 0;
 798         /* expiry */
 799         expiry = get_expiry(&mesg);
 800         status = -EINVAL;
 801         if (expiry == 0)
 802                 goto out;
 803
 804         /* remote flag */
 805         rv = get_int(&mesg, &tmp_int);
 806         if (rv) {
 807                 CERROR("fail to get remote flag\n");
 808                 goto out;
 809         }
 810         rsci.ctx.gsc_remote = (tmp_int != 0);
 811
 812         /* root user flag */
 813         rv = get_int(&mesg, &tmp_int);
 814         if (rv) {
 815                 CERROR("fail to get oss user flag\n");
 816                 goto out;
 817         }
 818         rsci.ctx.gsc_usr_root = (tmp_int != 0);
 819
 820         /* mds user flag */
 821         rv = get_int(&mesg, &tmp_int);
 822         if (rv) {
 823                 CERROR("fail to get mds user flag\n");
 824                 goto out;
 825         }
 826         rsci.ctx.gsc_usr_mds = (tmp_int != 0);
 827
 828         /* mapped uid */
 829         rv = get_int(&mesg, (int *) &rsci.ctx.gsc_mapped_uid);
 830         if (rv) {
 831                 CERROR("fail to get mapped uid\n");
 832                 goto out;
 833         }
 834
 835         /* uid, or NEGATIVE */
 836         rv = get_int(&mesg, (int *) &rsci.ctx.gsc_uid);
 837         if (rv == -EINVAL)
 838                 goto out;
 839         if (rv == -ENOENT) {
 840                 CERROR("NOENT? set rsc entry negative\n");
 841                 set_bit(CACHE_NEGATIVE, &rsci.h.flags);
 842         } else {
 843                 struct gss_api_mech *gm;
 844                 rawobj_t tmp_buf;
 845                 unsigned long ctx_expiry;
 846
 847                 /* gid */
 848                 if (get_int(&mesg, (int *) &rsci.ctx.gsc_gid))
 849                         goto out;
 850
 851                 /* mech name */
 852                 len = qword_get(&mesg, buf, mlen);
 853                 if (len < 0)
 854                         goto out;
 855                 gm = lgss_name_to_mech(buf);
 856                 status = -EOPNOTSUPP;
 857                 if (!gm)
 858                         goto out;
 859
 860                 status = -EINVAL;
 861                 /* mech-specific data: */
 862                 len = qword_get(&mesg, buf, mlen);
 863                 if (len < 0) {
 864                         lgss_mech_put(gm);
 865                         goto out;
 866                 }
 867                 tmp_buf.len = len;
 868                 tmp_buf.data = (unsigned char *)buf;
 869                 if (lgss_import_sec_context(&tmp_buf, gm,
 870                                             &rsci.ctx.gsc_mechctx)) {
 871                         lgss_mech_put(gm);
 872                         goto out;
 873                 }
 874
 875                 /* currently the expiry time passed down from user-space
 876                  * is invalid, here we retrive it from mech. */
 877                 if (lgss_inquire_context(rsci.ctx.gsc_mechctx, &ctx_expiry)) {
 878                         CERROR("unable to get expire time, drop it\n");
 879                         lgss_mech_put(gm);
 880                         goto out;
 881                 }
 882                 expiry = (time_t) ctx_expiry;
 883
 884                 lgss_mech_put(gm);
 885         }
 886
 887         rsci.h.expiry_time = expiry;
 888         rscp = rsc_lookup(&rsci, 1);
 889         status = 0;
 890 out:
 891         rsc_free(&rsci);
 892         if (rscp)
 893                 rsc_put(&rscp->h, &rsc_cache);
 894
 895         if (status)
 896                 CERROR("parse rsc error %d\n", status);
 897         return status;
 898 }
 899
 900 #endif /* HAVE_SUNRPC_CACHE_V2 */
 901
 902
 903 static struct cache_detail rsc_cache = {
 904         .hash_size      = RSC_HASHMAX,
 905         .hash_table     = rsc_table,
 906         .name           = "auth.sptlrpc.context",
 907         .cache_put      = rsc_put,
 908         .cache_parse    = rsc_parse,
 909 #ifdef HAVE_SUNRPC_CACHE_V2
 910         .match          = rsc_match,
 911         .init           = rsc_init,
 912         .update         = update_rsc,
 913         .alloc          = rsc_alloc,
 914 #endif
 915 };
 916
 917 #ifdef HAVE_SUNRPC_CACHE_V2
 918
 919 static struct rsc *rsc_lookup(struct rsc *item)
 920 {
 921         struct cache_head *ch;
 922         int                hash = rsc_hash(item);
 923
 924         ch = sunrpc_cache_lookup(&rsc_cache, &item->h, hash);
 925         if (ch)
 926                 return container_of(ch, struct rsc, h);
 927         else
 928                 return NULL;
 929 }
 930
 931 static struct rsc *rsc_update(struct rsc *new, struct rsc *old)
 932 {
 933         struct cache_head *ch;
 934         int                hash = rsc_hash(new);
 935
 936         ch = sunrpc_cache_update(&rsc_cache, &new->h, &old->h, hash);
 937         if (ch)
 938                 return container_of(ch, struct rsc, h);
 939         else
 940                 return NULL;
 941 }
 942
 943 #define COMPAT_RSC_PUT(item, cd)        cache_put((item), (cd))
 944
 945 #else
 946
 947 static DefineSimpleCacheLookup(rsc, 0);
 948
 949 #define COMPAT_RSC_PUT(item, cd)        rsc_put((item), (cd))
 950
 951 #endif
 952
 953 /****************************************
 954  * rsc cache flush                      *
 955  ****************************************/
 956
 957 typedef int rsc_entry_match(struct rsc *rscp, long data);
 958
 959 static void rsc_flush(rsc_entry_match *match, long data)
 960 {
 961         struct cache_head **ch;
 962         struct rsc *rscp;
 963         int n;
 964         ENTRY;
 965
 966         write_lock(&rsc_cache.hash_lock);
 967         for (n = 0; n < RSC_HASHMAX; n++) {
 968                 for (ch = &rsc_cache.hash_table[n]; *ch;) {
 969                         rscp = container_of(*ch, struct rsc, h);
 970
 971                         if (!match(rscp, data)) {
 972                                 ch = &((*ch)->next);
 973                                 continue;
 974                         }
 975
 976                         /* it seems simply set NEGATIVE doesn't work */
 977                         *ch = (*ch)->next;
 978                         rscp->h.next = NULL;
 979                         cache_get(&rscp->h);
 980                         set_bit(CACHE_NEGATIVE, &rscp->h.flags);
 981                         COMPAT_RSC_PUT(&rscp->h, &rsc_cache);
 982                         rsc_cache.entries--;
 983                 }
 984         }
 985         write_unlock(&rsc_cache.hash_lock);
 986         EXIT;
 987 }
 988
 989 static int match_uid(struct rsc *rscp, long uid)
 990 {
 991         if ((int) uid == -1)
 992                 return 1;
 993         return ((int) rscp->ctx.gsc_uid == (int) uid);
 994 }
 995
 996 static int match_target(struct rsc *rscp, long target)
 997 {
 998         return (rscp->target == (struct obd_device *) target);
 999 }
1000
1001 static inline void rsc_flush_uid(int uid)
1002 {
1003         if (uid == -1)
1004                 CWARN("flush all gss contexts...\n");
1005
1006         rsc_flush(match_uid, (long) uid);
1007 }
1008
1009 static inline void rsc_flush_target(struct obd_device *target)
1010 {
1011         rsc_flush(match_target, (long) target);
1012 }
1013
1014 void gss_secsvc_flush(struct obd_device *target)
1015 {
1016         rsc_flush_target(target);
1017 }
1018 EXPORT_SYMBOL(gss_secsvc_flush);
1019
1020 static struct rsc *gss_svc_searchbyctx(rawobj_t *handle)
1021 {
1022         struct rsc  rsci;
1023         struct rsc *found;
1024
1025         memset(&rsci, 0, sizeof(rsci));
1026         if (rawobj_dup(&rsci.handle, handle))
1027                 return NULL;
1028
1029 #ifdef HAVE_SUNRPC_CACHE_V2
1030         found = rsc_lookup(&rsci);
1031 #else
1032         found = rsc_lookup(&rsci, 0);
1033 #endif
1034         rsc_free(&rsci);
1035         if (!found)
1036                 return NULL;
1037         if (cache_check(&rsc_cache, &found->h, NULL))
1038                 return NULL;
1039         return found;
1040 }
1041
1042 #ifdef HAVE_SUNRPC_CACHE_V2
1043
1044 int gss_svc_upcall_install_rvs_ctx(struct obd_import *imp,
1045                                    struct gss_sec *gsec,
1046                                    struct gss_cli_ctx *gctx)
1047 {
1048         struct rsc      rsci, *rscp = NULL;
1049         unsigned long   ctx_expiry;
1050         __u32           major;
1051         int             rc;
1052         ENTRY;
1053
1054         memset(&rsci, 0, sizeof(rsci));
1055
1056         if (rawobj_alloc(&rsci.handle, (char *) &gsec->gs_rvs_hdl,
1057                          sizeof(gsec->gs_rvs_hdl)))
1058                 GOTO(out, rc = -ENOMEM);
1059
1060         rscp = rsc_lookup(&rsci);
1061         if (rscp == NULL)
1062                 GOTO(out, rc = -ENOMEM);
1063
1064         major = lgss_copy_reverse_context(gctx->gc_mechctx,
1065                                           &rsci.ctx.gsc_mechctx);
1066         if (major != GSS_S_COMPLETE)
1067                 GOTO(out, rc = -ENOMEM);
1068
1069         if (lgss_inquire_context(rsci.ctx.gsc_mechctx, &ctx_expiry)) {
1070                 CERROR("unable to get expire time, drop it\n");
1071                 GOTO(out, rc = -EINVAL);
1072         }
1073         rsci.h.expiry_time = (time_t) ctx_expiry;
1074
1075         /* FIXME */
1076         rsci.ctx.gsc_usr_root = 1;
1077         rsci.ctx.gsc_usr_mds= 1;
1078         rsci.ctx.gsc_reverse = 1;
1079
1080         rscp = rsc_update(&rsci, rscp);
1081         if (rscp == NULL)
1082                 GOTO(out, rc = -ENOMEM);
1083
1084         rscp->target = imp->imp_obd;
1085         rawobj_dup(&gctx->gc_svc_handle, &rscp->handle);
1086
1087         CWARN("create reverse svc ctx %p to %s: idx "LPX64"\n",
1088               &rscp->ctx, obd2cli_tgt(imp->imp_obd), gsec->gs_rvs_hdl);
1089         rc = 0;
1090 out:
1091         if (rscp)
1092                 cache_put(&rscp->h, &rsc_cache);
1093         rsc_free(&rsci);
1094
1095         if (rc)
1096                 CERROR("create reverse svc ctx: idx "LPX64", rc %d\n",
1097                        gsec->gs_rvs_hdl, rc);
1098         RETURN(rc);
1099 }
1100
1101 #else /* !HAVE_SUNRPC_CACHE_V2 */
1102
1103 int gss_svc_upcall_install_rvs_ctx(struct obd_import *imp,
1104                                    struct gss_sec *gsec,
1105                                    struct gss_cli_ctx *gctx)
1106 {
1107         struct rsc      rsci, *rscp;
1108         unsigned long   ctx_expiry;
1109         __u32           major;
1110         int             rc;
1111         ENTRY;
1112
1113         memset(&rsci, 0, sizeof(rsci));
1114
1115         if (rawobj_alloc(&rsci.handle, (char *) &gsec->gs_rvs_hdl,
1116                          sizeof(gsec->gs_rvs_hdl)))
1117                 GOTO(out, rc = -ENOMEM);
1118
1119         major = lgss_copy_reverse_context(gctx->gc_mechctx,
1120                                           &rsci.ctx.gsc_mechctx);
1121         if (major != GSS_S_COMPLETE)
1122                 GOTO(out, rc = -ENOMEM);
1123
1124         if (lgss_inquire_context(rsci.ctx.gsc_mechctx, &ctx_expiry)) {
1125                 CERROR("unable to get expire time, drop it\n");
1126                 GOTO(out, rc = -ENOMEM);
1127         }
1128         rsci.h.expiry_time = (time_t) ctx_expiry;
1129
1130         /* FIXME */
1131         rsci.ctx.gsc_usr_root = 1;
1132         rsci.ctx.gsc_usr_mds= 1;
1133         rsci.ctx.gsc_reverse = 1;
1134
1135         rscp = rsc_lookup(&rsci, 1);
1136         if (rscp == NULL) {
1137                 CERROR("rsc lookup failed\n");
1138                 GOTO(out, rc = -ENOMEM);
1139         }
1140
1141         rscp->target = imp->imp_obd;
1142         rawobj_dup(&gctx->gc_svc_handle, &rscp->handle);
1143
1144         CWARN("create reverse svc ctx %p to %s: idx "LPX64"\n",
1145               &rscp->ctx, obd2cli_tgt(imp->imp_obd), gsec->gs_rvs_hdl);
1146         rsc_put(&rscp->h, &rsc_cache);
1147         rc = 0;
1148 out:
1149         rsc_free(&rsci);
1150         if (rc)
1151                 CERROR("create reverse svc ctx: idx "LPX64", rc %d\n",
1152                        gsec->gs_rvs_hdl, rc);
1153         RETURN(rc);
1154 }
1155
1156 #endif /* HAVE_SUNRPC_CACHE_V2 */
1157
1158 int gss_svc_upcall_expire_rvs_ctx(rawobj_t *handle)
1159 {
1160         const cfs_time_t        expire = 20;
1161         struct rsc             *rscp;
1162
1163         rscp = gss_svc_searchbyctx(handle);
1164         if (rscp) {
1165                 CDEBUG(D_SEC, "reverse svcctx %p (rsc %p) expire soon\n",
1166                        &rscp->ctx, rscp);
1167
1168                 rscp->h.expiry_time = cfs_time_current_sec() + expire;
1169                 COMPAT_RSC_PUT(&rscp->h, &rsc_cache);
1170         }
1171         return 0;
1172 }
1173
1174 int gss_svc_upcall_dup_handle(rawobj_t *handle, struct gss_svc_ctx *ctx)
1175 {
1176         struct rsc *rscp = container_of(ctx, struct rsc, ctx);
1177
1178         return rawobj_dup(handle, &rscp->handle);
1179 }
1180
1181 int gss_svc_upcall_update_sequence(rawobj_t *handle, __u32 seq)
1182 {
1183         struct rsc             *rscp;
1184
1185         rscp = gss_svc_searchbyctx(handle);
1186         if (rscp) {
1187                 CDEBUG(D_SEC, "reverse svcctx %p (rsc %p) update seq to %u\n",
1188                        &rscp->ctx, rscp, seq + 1);
1189
1190                 rscp->ctx.gsc_rvs_seq = seq + 1;
1191                 COMPAT_RSC_PUT(&rscp->h, &rsc_cache);
1192         }
1193         return 0;
1194 }
1195
1196 static struct cache_deferred_req* cache_upcall_defer(struct cache_req *req)
1197 {
1198         return NULL;
1199 }
1200 static struct cache_req cache_upcall_chandle = { cache_upcall_defer };
1201
1202 int gss_svc_upcall_handle_init(struct ptlrpc_request *req,
1203                                struct gss_svc_reqctx *grctx,
1204                                struct gss_wire_ctx *gw,
1205                                struct obd_device *target,
1206                                __u32 lustre_svc,
1207                                rawobj_t *rvs_hdl,
1208                                rawobj_t *in_token)
1209 {
1210         struct ptlrpc_reply_state *rs;
1211         struct rsc                *rsci = NULL;
1212         struct rsi                *rsip = NULL, rsikey;
1213         wait_queue_t               wait;
1214         int                        replen = sizeof(struct ptlrpc_body);
1215         struct gss_rep_header     *rephdr;
1216         int                        first_check = 1;
1217         int                        rc = SECSVC_DROP;
1218         ENTRY;
1219
1220         memset(&rsikey, 0, sizeof(rsikey));
1221         rsikey.lustre_svc = lustre_svc;
1222         rsikey.nid = (__u64) req->rq_peer.nid;
1223
1224         /* duplicate context handle. for INIT it always 0 */
1225         if (rawobj_dup(&rsikey.in_handle, &gw->gw_handle)) {
1226                 CERROR("fail to dup context handle\n");
1227                 GOTO(out, rc);
1228         }
1229
1230         if (rawobj_dup(&rsikey.in_token, in_token)) {
1231                 CERROR("can't duplicate token\n");
1232                 rawobj_free(&rsikey.in_handle);
1233                 GOTO(out, rc);
1234         }
1235
1236 #ifdef HAVE_SUNRPC_CACHE_V2
1237         rsip = rsi_lookup(&rsikey);
1238 #else
1239         rsip = rsi_lookup(&rsikey, 0);
1240 #endif
1241         rsi_free(&rsikey);
1242         if (!rsip) {
1243                 CERROR("error in rsi_lookup.\n");
1244
1245                 if (!gss_pack_err_notify(req, GSS_S_FAILURE, 0))
1246                         rc = SECSVC_COMPLETE;
1247
1248                 GOTO(out, rc);
1249         }
1250
1251         cache_get(&rsip->h); /* take an extra ref */
1252         init_waitqueue_head(&rsip->waitq);
1253         init_waitqueue_entry(&wait, current);
1254         add_wait_queue(&rsip->waitq, &wait);
1255
1256 cache_check:
1257         /* Note each time cache_check() will drop a reference if return
1258          * non-zero. We hold an extra reference on initial rsip, but must
1259          * take care of following calls. */
1260         rc = cache_check(&rsi_cache, &rsip->h, &cache_upcall_chandle);
1261         switch (rc) {
1262         case -EAGAIN: {
1263                 int valid;
1264
1265                 if (first_check) {
1266                         first_check = 0;
1267
1268                         read_lock(&rsi_cache.hash_lock);
1269                         valid = test_bit(CACHE_VALID, &rsip->h.flags);
1270                         if (valid == 0)
1271                                 set_current_state(TASK_INTERRUPTIBLE);
1272                         read_unlock(&rsi_cache.hash_lock);
1273
1274                         if (valid == 0)
1275                                 schedule_timeout(GSS_SVC_UPCALL_TIMEOUT * HZ);
1276
1277                         cache_get(&rsip->h);
1278                         goto cache_check;
1279                 }
1280                 CWARN("waited %ds timeout, drop\n", GSS_SVC_UPCALL_TIMEOUT);
1281                 break;
1282         }
1283         case -ENOENT:
1284                 CWARN("cache_check return ENOENT, drop\n");
1285                 break;
1286         case 0:
1287                 /* if not the first check, we have to release the extra
1288                  * reference we just added on it. */
1289                 if (!first_check)
1290                         cache_put(&rsip->h, &rsi_cache);
1291                 CDEBUG(D_SEC, "cache_check is good\n");
1292                 break;
1293         }
1294
1295         remove_wait_queue(&rsip->waitq, &wait);
1296         cache_put(&rsip->h, &rsi_cache);
1297
1298         if (rc)
1299                 GOTO(out, rc = SECSVC_DROP);
1300
1301         rc = SECSVC_DROP;
1302         rsci = gss_svc_searchbyctx(&rsip->out_handle);
1303         if (!rsci) {
1304                 CERROR("authentication failed\n");
1305
1306                 if (!gss_pack_err_notify(req, GSS_S_FAILURE, 0))
1307                         rc = SECSVC_COMPLETE;
1308
1309                 GOTO(out, rc);
1310         } else {
1311                 cache_get(&rsci->h);
1312                 grctx->src_ctx = &rsci->ctx;
1313         }
1314
1315         if (rawobj_dup(&rsci->ctx.gsc_rvs_hdl, rvs_hdl)) {
1316                 CERROR("failed duplicate reverse handle\n");
1317                 GOTO(out, rc);
1318         }
1319
1320         rsci->target = target;
1321
1322         CDEBUG(D_SEC, "server create rsc %p(%u->%s)\n",
1323                rsci, rsci->ctx.gsc_uid, libcfs_nid2str(req->rq_peer.nid));
1324
1325         if (rsip->out_handle.len > PTLRPC_GSS_MAX_HANDLE_SIZE) {
1326                 CERROR("handle size %u too large\n", rsip->out_handle.len);
1327                 GOTO(out, rc = SECSVC_DROP);
1328         }
1329
1330         grctx->src_init = 1;
1331         grctx->src_reserve_len = size_round4(rsip->out_token.len);
1332
1333         rc = lustre_pack_reply_v2(req, 1, &replen, NULL);
1334         if (rc) {
1335                 CERROR("failed to pack reply: %d\n", rc);
1336                 GOTO(out, rc = SECSVC_DROP);
1337         }
1338
1339         rs = req->rq_reply_state;
1340         LASSERT(rs->rs_repbuf->lm_bufcount == 3);
1341         LASSERT(rs->rs_repbuf->lm_buflens[0] >=
1342                 sizeof(*rephdr) + rsip->out_handle.len);
1343         LASSERT(rs->rs_repbuf->lm_buflens[2] >= rsip->out_token.len);
1344
1345         rephdr = lustre_msg_buf(rs->rs_repbuf, 0, 0);
1346         rephdr->gh_version = PTLRPC_GSS_VERSION;
1347         rephdr->gh_flags = 0;
1348         rephdr->gh_proc = PTLRPC_GSS_PROC_ERR;
1349         rephdr->gh_major = rsip->major_status;
1350         rephdr->gh_minor = rsip->minor_status;
1351         rephdr->gh_seqwin = GSS_SEQ_WIN;
1352         rephdr->gh_handle.len = rsip->out_handle.len;
1353         memcpy(rephdr->gh_handle.data, rsip->out_handle.data,
1354                rsip->out_handle.len);
1355
1356         memcpy(lustre_msg_buf(rs->rs_repbuf, 2, 0), rsip->out_token.data,
1357                rsip->out_token.len);
1358
1359         rs->rs_repdata_len = lustre_shrink_msg(rs->rs_repbuf, 2,
1360                                                rsip->out_token.len, 0);
1361
1362         rc = SECSVC_OK;
1363
1364 out:
1365         /* it looks like here we should put rsip also, but this mess up
1366          * with NFS cache mgmt code... FIXME */
1367 #if 0
1368         if (rsip)
1369                 rsi_put(&rsip->h, &rsi_cache);
1370 #endif
1371
1372         if (rsci) {
1373                 /* if anything went wrong, we don't keep the context too */
1374                 if (rc != SECSVC_OK)
1375                         set_bit(CACHE_NEGATIVE, &rsci->h.flags);
1376                 else
1377                         CDEBUG(D_SEC, "create rsc with idx "LPX64"\n",
1378                                gss_handle_to_u64(&rsci->handle));
1379
1380                 COMPAT_RSC_PUT(&rsci->h, &rsc_cache);
1381         }
1382         RETURN(rc);
1383 }
1384
1385 struct gss_svc_ctx *gss_svc_upcall_get_ctx(struct ptlrpc_request *req,
1386                                            struct gss_wire_ctx *gw)
1387 {
1388         struct rsc *rsc;
1389
1390         rsc = gss_svc_searchbyctx(&gw->gw_handle);
1391         if (!rsc) {
1392                 CWARN("Invalid gss ctx idx "LPX64" from %s\n",
1393                       gss_handle_to_u64(&gw->gw_handle),
1394                       libcfs_nid2str(req->rq_peer.nid));
1395                 return NULL;
1396         }
1397
1398         return &rsc->ctx;
1399 }
1400
1401 void gss_svc_upcall_put_ctx(struct gss_svc_ctx *ctx)
1402 {
1403         struct rsc *rsc = container_of(ctx, struct rsc, ctx);
1404
1405         COMPAT_RSC_PUT(&rsc->h, &rsc_cache);
1406 }
1407
1408 void gss_svc_upcall_destroy_ctx(struct gss_svc_ctx *ctx)
1409 {
1410         struct rsc *rsc = container_of(ctx, struct rsc, ctx);
1411
1412         /* can't be found */
1413         set_bit(CACHE_NEGATIVE, &rsc->h.flags);
1414         /* to be removed at next scan */
1415         rsc->h.expiry_time = 1;
1416 }
1417
1418 int __init gss_init_svc_upcall(void)
1419 {
1420         int     i;
1421
1422         cache_register(&rsi_cache);
1423         cache_register(&rsc_cache);
1424
1425         /* FIXME this looks stupid. we intend to give lsvcgssd a chance to open
1426          * the init upcall channel, otherwise there's big chance that the first
1427          * upcall issued before the channel be opened thus nfsv4 cache code will
1428          * drop the request direclty, thus lead to unnecessary recovery time.
1429          * here we wait at miximum 1.5 seconds. */
1430         for (i = 0; i < 6; i++) {
1431                 if (atomic_read(&rsi_cache.readers) > 0)
1432                         break;
1433                 set_current_state(TASK_UNINTERRUPTIBLE);
1434                 LASSERT(HZ >= 4);
1435                 schedule_timeout(HZ / 4);
1436         }
1437
1438         if (atomic_read(&rsi_cache.readers) == 0)
1439                 CWARN("Init channel is not opened by lsvcgssd, following "
1440                       "request might be dropped until lsvcgssd is active\n");
1441
1442         /* this helps reducing context index confliction. after server reboot,
1443          * conflicting request from clients might be filtered out by initial
1444          * sequence number checking, thus no chance to sent error notification
1445          * back to clients. */
1446         get_random_bytes(&__ctx_index, sizeof(__ctx_index));
1447
1448         return 0;
1449 }
1450
1451 void __exit gss_exit_svc_upcall(void)
1452 {
1453         int rc;
1454
1455         cache_purge(&rsi_cache);
1456         if ((rc = cache_unregister(&rsi_cache)))
1457                 CERROR("unregister rsi cache: %d\n", rc);
1458
1459         cache_purge(&rsc_cache);
1460         if ((rc = cache_unregister(&rsc_cache)))
1461                 CERROR("unregister rsc cache: %d\n", rc);
1462 }