View Javadoc

1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  
19  package org.apache.hadoop.hbase.security.access;
20  
21  import java.io.IOException;
22  import java.util.HashMap;
23  import java.util.List;
24  import java.util.Map;
25  import java.util.concurrent.ConcurrentSkipListMap;
26  
27  import org.apache.commons.logging.Log;
28  import org.apache.commons.logging.LogFactory;
29  import org.apache.hadoop.hbase.classification.InterfaceAudience;
30  import org.apache.hadoop.conf.Configuration;
31  import org.apache.hadoop.hbase.Cell;
32  import org.apache.hadoop.hbase.TableName;
33  import org.apache.hadoop.hbase.exceptions.DeserializationException;
34  import org.apache.hadoop.hbase.security.User;
35  import org.apache.hadoop.hbase.security.UserProvider;
36  import org.apache.hadoop.hbase.util.Bytes;
37  import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
38  import org.apache.zookeeper.KeeperException;
39  
40  import com.google.common.collect.ArrayListMultimap;
41  import com.google.common.collect.ListMultimap;
42  import com.google.common.collect.Lists;
43  
44  /**
45   * Performs authorization checks for a given user's assigned permissions
46   */
47  @InterfaceAudience.Private
48  public class TableAuthManager {
49    private static class PermissionCache<T extends Permission> {
50      /** Cache of user permissions */
51      private ListMultimap<String,T> userCache = ArrayListMultimap.create();
52      /** Cache of group permissions */
53      private ListMultimap<String,T> groupCache = ArrayListMultimap.create();
54  
55      public List<T> getUser(String user) {
56        return userCache.get(user);
57      }
58  
59      public void putUser(String user, T perm) {
60        userCache.put(user, perm);
61      }
62  
63      public List<T> replaceUser(String user, Iterable<? extends T> perms) {
64        return userCache.replaceValues(user, perms);
65      }
66  
67      public List<T> getGroup(String group) {
68        return groupCache.get(group);
69      }
70  
71      public void putGroup(String group, T perm) {
72        groupCache.put(group, perm);
73      }
74  
75      public List<T> replaceGroup(String group, Iterable<? extends T> perms) {
76        return groupCache.replaceValues(group, perms);
77      }
78  
79      /**
80       * Returns a combined map of user and group permissions, with group names prefixed by
81       * {@link AccessControlLists#GROUP_PREFIX}.
82       */
83      public ListMultimap<String,T> getAllPermissions() {
84        ListMultimap<String,T> tmp = ArrayListMultimap.create();
85        tmp.putAll(userCache);
86        for (String group : groupCache.keySet()) {
87          tmp.putAll(AccessControlLists.GROUP_PREFIX + group, groupCache.get(group));
88        }
89        return tmp;
90      }
91    }
92  
93    private static Log LOG = LogFactory.getLog(TableAuthManager.class);
94  
95    private static TableAuthManager instance;
96  
97    /** Cache of global permissions */
98    private volatile PermissionCache<Permission> globalCache;
99  
100   private ConcurrentSkipListMap<TableName, PermissionCache<TablePermission>> tableCache =
101       new ConcurrentSkipListMap<TableName, PermissionCache<TablePermission>>();
102 
103   private ConcurrentSkipListMap<String, PermissionCache<TablePermission>> nsCache =
104     new ConcurrentSkipListMap<String, PermissionCache<TablePermission>>();
105 
106   private Configuration conf;
107   private ZKPermissionWatcher zkperms;
108   private volatile long mtime;
109 
110   private TableAuthManager(ZooKeeperWatcher watcher, Configuration conf)
111       throws IOException {
112     this.conf = conf;
113 
114     // initialize global permissions based on configuration
115     globalCache = initGlobal(conf);
116 
117     this.zkperms = new ZKPermissionWatcher(watcher, this, conf);
118     try {
119       this.zkperms.start();
120     } catch (KeeperException ke) {
121       LOG.error("ZooKeeper initialization failed", ke);
122     }
123   }
124 
125   /**
126    * Returns a new {@code PermissionCache} initialized with permission assignments
127    * from the {@code hbase.superuser} configuration key.
128    */
129   private PermissionCache<Permission> initGlobal(Configuration conf) throws IOException {
130     UserProvider userProvider = UserProvider.instantiate(conf);
131     User user = userProvider.getCurrent();
132     if (user == null) {
133       throw new IOException("Unable to obtain the current user, " +
134           "authorization checks for internal operations will not work correctly!");
135     }
136     PermissionCache<Permission> newCache = new PermissionCache<Permission>();
137     String currentUser = user.getShortName();
138 
139     // the system user is always included
140     List<String> superusers = Lists.asList(currentUser, conf.getStrings(
141         AccessControlLists.SUPERUSER_CONF_KEY, new String[0]));
142     if (superusers != null) {
143       for (String name : superusers) {
144         if (AccessControlLists.isGroupPrincipal(name)) {
145           newCache.putGroup(AccessControlLists.getGroupName(name),
146               new Permission(Permission.Action.values()));
147         } else {
148           newCache.putUser(name, new Permission(Permission.Action.values()));
149         }
150       }
151     }
152     return newCache;
153   }
154 
155   public ZKPermissionWatcher getZKPermissionWatcher() {
156     return this.zkperms;
157   }
158 
159   public void refreshTableCacheFromWritable(TableName table,
160                                        byte[] data) throws IOException {
161     if (data != null && data.length > 0) {
162       ListMultimap<String,TablePermission> perms;
163       try {
164         perms = AccessControlLists.readPermissions(data, conf);
165       } catch (DeserializationException e) {
166         throw new IOException(e);
167       }
168 
169       if (perms != null) {
170         if (Bytes.equals(table.getName(), AccessControlLists.ACL_GLOBAL_NAME)) {
171           updateGlobalCache(perms);
172         } else {
173           updateTableCache(table, perms);
174         }
175       }
176     } else {
177       LOG.debug("Skipping permission cache refresh because writable data is empty");
178     }
179   }
180 
181   public void refreshNamespaceCacheFromWritable(String namespace, byte[] data) throws IOException {
182     if (data != null && data.length > 0) {
183       ListMultimap<String,TablePermission> perms;
184       try {
185         perms = AccessControlLists.readPermissions(data, conf);
186       } catch (DeserializationException e) {
187         throw new IOException(e);
188       }
189       if (perms != null) {
190         updateNsCache(namespace, perms);
191       }
192     } else {
193       LOG.debug("Skipping permission cache refresh because writable data is empty");
194     }
195   }
196 
197   /**
198    * Updates the internal global permissions cache
199    *
200    * @param userPerms
201    */
202   private void updateGlobalCache(ListMultimap<String,TablePermission> userPerms) {
203     PermissionCache<Permission> newCache = null;
204     try {
205       newCache = initGlobal(conf);
206       for (Map.Entry<String,TablePermission> entry : userPerms.entries()) {
207         if (AccessControlLists.isGroupPrincipal(entry.getKey())) {
208           newCache.putGroup(AccessControlLists.getGroupName(entry.getKey()),
209               new Permission(entry.getValue().getActions()));
210         } else {
211           newCache.putUser(entry.getKey(), new Permission(entry.getValue().getActions()));
212         }
213       }
214       globalCache = newCache;
215       mtime++;
216     } catch (IOException e) {
217       // Never happens
218       LOG.error("Error occured while updating the global cache", e);
219     }
220   }
221 
222   /**
223    * Updates the internal permissions cache for a single table, splitting
224    * the permissions listed into separate caches for users and groups to optimize
225    * group lookups.
226    *
227    * @param table
228    * @param tablePerms
229    */
230   private void updateTableCache(TableName table,
231                                 ListMultimap<String,TablePermission> tablePerms) {
232     PermissionCache<TablePermission> newTablePerms = new PermissionCache<TablePermission>();
233 
234     for (Map.Entry<String,TablePermission> entry : tablePerms.entries()) {
235       if (AccessControlLists.isGroupPrincipal(entry.getKey())) {
236         newTablePerms.putGroup(AccessControlLists.getGroupName(entry.getKey()), entry.getValue());
237       } else {
238         newTablePerms.putUser(entry.getKey(), entry.getValue());
239       }
240     }
241 
242     tableCache.put(table, newTablePerms);
243     mtime++;
244   }
245 
246   /**
247    * Updates the internal permissions cache for a single table, splitting
248    * the permissions listed into separate caches for users and groups to optimize
249    * group lookups.
250    *
251    * @param namespace
252    * @param tablePerms
253    */
254   private void updateNsCache(String namespace,
255                              ListMultimap<String, TablePermission> tablePerms) {
256     PermissionCache<TablePermission> newTablePerms = new PermissionCache<TablePermission>();
257 
258     for (Map.Entry<String, TablePermission> entry : tablePerms.entries()) {
259       if (AccessControlLists.isGroupPrincipal(entry.getKey())) {
260         newTablePerms.putGroup(AccessControlLists.getGroupName(entry.getKey()), entry.getValue());
261       } else {
262         newTablePerms.putUser(entry.getKey(), entry.getValue());
263       }
264     }
265 
266     nsCache.put(namespace, newTablePerms);
267     mtime++;
268   }
269 
270   private PermissionCache<TablePermission> getTablePermissions(TableName table) {
271     if (!tableCache.containsKey(table)) {
272       tableCache.putIfAbsent(table, new PermissionCache<TablePermission>());
273     }
274     return tableCache.get(table);
275   }
276 
277   private PermissionCache<TablePermission> getNamespacePermissions(String namespace) {
278     if (!nsCache.containsKey(namespace)) {
279       nsCache.putIfAbsent(namespace, new PermissionCache<TablePermission>());
280     }
281     return nsCache.get(namespace);
282   }
283 
284   /**
285    * Authorizes a global permission
286    * @param perms
287    * @param action
288    * @return true if authorized, false otherwise
289    */
290   private boolean authorize(List<Permission> perms, Permission.Action action) {
291     if (perms != null) {
292       for (Permission p : perms) {
293         if (p.implies(action)) {
294           return true;
295         }
296       }
297     } else if (LOG.isDebugEnabled()) {
298       LOG.debug("No permissions found for " + action);
299     }
300 
301     return false;
302   }
303 
304   /**
305    * Authorize a global permission based on ACLs for the given user and the
306    * user's groups.
307    * @param user
308    * @param action
309    * @return true if known and authorized, false otherwise
310    */
311   public boolean authorize(User user, Permission.Action action) {
312     if (user == null) {
313       return false;
314     }
315 
316     if (authorize(globalCache.getUser(user.getShortName()), action)) {
317       return true;
318     }
319 
320     String[] groups = user.getGroupNames();
321     if (groups != null) {
322       for (String group : groups) {
323         if (authorize(globalCache.getGroup(group), action)) {
324           return true;
325         }
326       }
327     }
328     return false;
329   }
330 
331   private boolean authorize(List<TablePermission> perms,
332                             TableName table, byte[] family,
333                             Permission.Action action) {
334     return authorize(perms, table, family, null, action);
335   }
336 
337   private boolean authorize(List<TablePermission> perms,
338                             TableName table, byte[] family,
339                             byte[] qualifier, Permission.Action action) {
340     if (perms != null) {
341       for (TablePermission p : perms) {
342         if (p.implies(table, family, qualifier, action)) {
343           return true;
344         }
345       }
346     } else if (LOG.isDebugEnabled()) {
347       LOG.debug("No permissions found for table="+table);
348     }
349     return false;
350   }
351 
352   private boolean hasAccess(List<TablePermission> perms,
353                             TableName table, Permission.Action action) {
354     if (perms != null) {
355       for (TablePermission p : perms) {
356         if (p.implies(action)) {
357           return true;
358         }
359       }
360     } else if (LOG.isDebugEnabled()) {
361       LOG.debug("No permissions found for table="+table);
362     }
363     return false;
364   }
365 
366   /**
367    * Authorize a user for a given KV. This is called from AccessControlFilter.
368    */
369   public boolean authorize(User user, TableName table, Cell cell, Permission.Action action) {
370     try {
371       List<Permission> perms = AccessControlLists.getCellPermissionsForUser(user, cell);
372       if (LOG.isTraceEnabled()) {
373         LOG.trace("Perms for user " + user.getShortName() + " in cell " + cell + ": " +
374           (perms != null ? perms : ""));
375       }
376       if (perms != null) {
377         for (Permission p: perms) {
378           if (p.implies(action)) {
379             return true;
380           }
381         }
382       }
383     } catch (IOException e) {
384       // We failed to parse the KV tag
385       LOG.error("Failed parse of ACL tag in cell " + cell);
386       // Fall through to check with the table and CF perms we were able
387       // to collect regardless
388     }
389     return false;
390   }
391 
392   public boolean authorize(User user, String namespace, Permission.Action action) {
393     // Global authorizations supercede namespace level
394     if (authorizeUser(user, action)) {
395       return true;
396     }
397     // Check namespace permissions
398     PermissionCache<TablePermission> tablePerms = nsCache.get(namespace);
399     if (tablePerms != null) {
400       List<TablePermission> userPerms = tablePerms.getUser(user.getShortName());
401       if (authorize(userPerms, namespace, action)) {
402         return true;
403       }
404       String[] groupNames = user.getGroupNames();
405       if (groupNames != null) {
406         for (String group : groupNames) {
407           List<TablePermission> groupPerms = tablePerms.getGroup(group);
408           if (authorize(groupPerms, namespace, action)) {
409             return true;
410           }
411         }
412       }
413     }
414     return false;
415   }
416 
417   private boolean authorize(List<TablePermission> perms, String namespace,
418                             Permission.Action action) {
419     if (perms != null) {
420       for (TablePermission p : perms) {
421         if (p.implies(namespace, action)) {
422           return true;
423         }
424       }
425     } else if (LOG.isDebugEnabled()) {
426       LOG.debug("No permissions for authorize() check, table=" + namespace);
427     }
428 
429     return false;
430   }
431 
432   /**
433    * Checks global authorization for a specific action for a user, based on the
434    * stored user permissions.
435    */
436   public boolean authorizeUser(User user, Permission.Action action) {
437     return authorize(globalCache.getUser(user.getShortName()), action);
438   }
439 
440   /**
441    * Checks authorization to a given table and column family for a user, based on the
442    * stored user permissions.
443    *
444    * @param user
445    * @param table
446    * @param family
447    * @param action
448    * @return true if known and authorized, false otherwise
449    */
450   public boolean authorizeUser(User user, TableName table, byte[] family,
451       Permission.Action action) {
452     return authorizeUser(user, table, family, null, action);
453   }
454 
455   public boolean authorizeUser(User user, TableName table, byte[] family,
456       byte[] qualifier, Permission.Action action) {
457     if (table == null) table = AccessControlLists.ACL_TABLE_NAME;
458     // Global and namespace authorizations supercede table level
459     if (authorize(user, table.getNamespaceAsString(), action)) {
460       return true;
461     }
462     // Check table permissions
463     return authorize(getTablePermissions(table).getUser(user.getShortName()), table, family,
464         qualifier, action);
465   }
466 
467   /**
468    * Checks if the user has access to the full table or at least a family/qualifier
469    * for the specified action.
470    *
471    * @param user
472    * @param table
473    * @param action
474    * @return true if the user has access to the table, false otherwise
475    */
476   public boolean userHasAccess(User user, TableName table, Permission.Action action) {
477     if (table == null) table = AccessControlLists.ACL_TABLE_NAME;
478     // Global and namespace authorizations supercede table level
479     if (authorize(user, table.getNamespaceAsString(), action)) {
480       return true;
481     }
482     // Check table permissions
483     return hasAccess(getTablePermissions(table).getUser(user.getShortName()), table, action);
484   }
485 
486   /**
487    * Checks global authorization for a given action for a group, based on the stored
488    * permissions.
489    */
490   public boolean authorizeGroup(String groupName, Permission.Action action) {
491     List<Permission> perms = globalCache.getGroup(groupName);
492     if (LOG.isDebugEnabled()) {
493       LOG.debug("authorizing " + (perms != null && !perms.isEmpty() ? perms.get(0) : "") +
494         " for " + action);
495     }
496     return authorize(perms, action);
497   }
498 
499   /**
500    * Checks authorization to a given table, column family and column for a group, based
501    * on the stored permissions.
502    * @param groupName
503    * @param table
504    * @param family
505    * @param qualifier
506    * @param action
507    * @return true if known and authorized, false otherwise
508    */
509   public boolean authorizeGroup(String groupName, TableName table, byte[] family,
510       byte[] qualifier, Permission.Action action) {
511     // Global authorization supercedes table level
512     if (authorizeGroup(groupName, action)) {
513       return true;
514     }
515     if (table == null) table = AccessControlLists.ACL_TABLE_NAME;
516     // Namespace authorization supercedes table level
517     String namespace = table.getNamespaceAsString();
518     if (authorize(getNamespacePermissions(namespace).getGroup(groupName), namespace, action)) {
519       return true;
520     }
521     // Check table level
522     List<TablePermission> tblPerms = getTablePermissions(table).getGroup(groupName);
523     if (LOG.isDebugEnabled()) {
524       LOG.debug("authorizing " + (tblPerms != null && !tblPerms.isEmpty() ? tblPerms.get(0) : "") +
525         " for " +groupName + " on " + table + "." + Bytes.toString(family) + "." +
526         Bytes.toString(qualifier) + " with " + action);
527     }
528     return authorize(tblPerms, table, family, qualifier, action);
529   }
530 
531   /**
532    * Checks if the user has access to the full table or at least a family/qualifier
533    * for the specified action.
534    * @param groupName
535    * @param table
536    * @param action
537    * @return true if the group has access to the table, false otherwise
538    */
539   public boolean groupHasAccess(String groupName, TableName table, Permission.Action action) {
540     // Global authorization supercedes table level
541     if (authorizeGroup(groupName, action)) {
542       return true;
543     }
544     if (table == null) table = AccessControlLists.ACL_TABLE_NAME;
545     // Namespace authorization supercedes table level
546     if (hasAccess(getNamespacePermissions(table.getNamespaceAsString()).getGroup(groupName),
547         table, action)) {
548       return true;
549     }
550     // Check table level
551     return hasAccess(getTablePermissions(table).getGroup(groupName), table, action);
552   }
553 
554   public boolean authorize(User user, TableName table, byte[] family,
555       byte[] qualifier, Permission.Action action) {
556     if (authorizeUser(user, table, family, qualifier, action)) {
557       return true;
558     }
559 
560     String[] groups = user.getGroupNames();
561     if (groups != null) {
562       for (String group : groups) {
563         if (authorizeGroup(group, table, family, qualifier, action)) {
564           return true;
565         }
566       }
567     }
568     return false;
569   }
570 
571   public boolean hasAccess(User user, TableName table, Permission.Action action) {
572     if (userHasAccess(user, table, action)) {
573       return true;
574     }
575 
576     String[] groups = user.getGroupNames();
577     if (groups != null) {
578       for (String group : groups) {
579         if (groupHasAccess(group, table, action)) {
580           return true;
581         }
582       }
583     }
584     return false;
585   }
586 
587   public boolean authorize(User user, TableName table, byte[] family,
588       Permission.Action action) {
589     return authorize(user, table, family, null, action);
590   }
591 
592   /**
593    * Returns true if the given user has a {@link TablePermission} matching up
594    * to the column family portion of a permission.  Note that this permission
595    * may be scoped to a given column qualifier and does not guarantee that
596    * authorize() on the same column family would return true.
597    */
598   public boolean matchPermission(User user,
599       TableName table, byte[] family, Permission.Action action) {
600     PermissionCache<TablePermission> tablePerms = tableCache.get(table);
601     if (tablePerms != null) {
602       List<TablePermission> userPerms = tablePerms.getUser(user.getShortName());
603       if (userPerms != null) {
604         for (TablePermission p : userPerms) {
605           if (p.matchesFamily(table, family, action)) {
606             return true;
607           }
608         }
609       }
610 
611       String[] groups = user.getGroupNames();
612       if (groups != null) {
613         for (String group : groups) {
614           List<TablePermission> groupPerms = tablePerms.getGroup(group);
615           if (groupPerms != null) {
616             for (TablePermission p : groupPerms) {
617               if (p.matchesFamily(table, family, action)) {
618                 return true;
619               }
620             }
621           }
622         }
623       }
624     }
625 
626     return false;
627   }
628 
629   public boolean matchPermission(User user,
630       TableName table, byte[] family, byte[] qualifier,
631       Permission.Action action) {
632     PermissionCache<TablePermission> tablePerms = tableCache.get(table);
633     if (tablePerms != null) {
634       List<TablePermission> userPerms = tablePerms.getUser(user.getShortName());
635       if (userPerms != null) {
636         for (TablePermission p : userPerms) {
637           if (p.matchesFamilyQualifier(table, family, qualifier, action)) {
638             return true;
639           }
640         }
641       }
642 
643       String[] groups = user.getGroupNames();
644       if (groups != null) {
645         for (String group : groups) {
646           List<TablePermission> groupPerms = tablePerms.getGroup(group);
647           if (groupPerms != null) {
648             for (TablePermission p : groupPerms) {
649               if (p.matchesFamilyQualifier(table, family, qualifier, action)) {
650                 return true;
651               }
652             }
653           }
654         }
655       }
656     }
657     return false;
658   }
659 
660   public void removeNamespace(byte[] ns) {
661     nsCache.remove(Bytes.toString(ns));
662   }
663 
664   public void removeTable(TableName table) {
665     tableCache.remove(table);
666   }
667 
668   /**
669    * Overwrites the existing permission set for a given user for a table, and
670    * triggers an update for zookeeper synchronization.
671    * @param username
672    * @param table
673    * @param perms
674    */
675   public void setTableUserPermissions(String username, TableName table,
676       List<TablePermission> perms) {
677     PermissionCache<TablePermission> tablePerms = getTablePermissions(table);
678     tablePerms.replaceUser(username, perms);
679     writeTableToZooKeeper(table, tablePerms);
680   }
681 
682   /**
683    * Overwrites the existing permission set for a group and triggers an update
684    * for zookeeper synchronization.
685    * @param group
686    * @param table
687    * @param perms
688    */
689   public void setTableGroupPermissions(String group, TableName table,
690       List<TablePermission> perms) {
691     PermissionCache<TablePermission> tablePerms = getTablePermissions(table);
692     tablePerms.replaceGroup(group, perms);
693     writeTableToZooKeeper(table, tablePerms);
694   }
695 
696   /**
697    * Overwrites the existing permission set for a given user for a table, and
698    * triggers an update for zookeeper synchronization.
699    * @param username
700    * @param namespace
701    * @param perms
702    */
703   public void setNamespaceUserPermissions(String username, String namespace,
704       List<TablePermission> perms) {
705     PermissionCache<TablePermission> tablePerms = getNamespacePermissions(namespace);
706     tablePerms.replaceUser(username, perms);
707     writeNamespaceToZooKeeper(namespace, tablePerms);
708   }
709 
710   /**
711    * Overwrites the existing permission set for a group and triggers an update
712    * for zookeeper synchronization.
713    * @param group
714    * @param namespace
715    * @param perms
716    */
717   public void setNamespaceGroupPermissions(String group, String namespace,
718       List<TablePermission> perms) {
719     PermissionCache<TablePermission> tablePerms = getNamespacePermissions(namespace);
720     tablePerms.replaceGroup(group, perms);
721     writeNamespaceToZooKeeper(namespace, tablePerms);
722   }
723 
724   public void writeTableToZooKeeper(TableName table,
725       PermissionCache<TablePermission> tablePerms) {
726     byte[] serialized = new byte[0];
727     if (tablePerms != null) {
728       serialized = AccessControlLists.writePermissionsAsBytes(tablePerms.getAllPermissions(), conf);
729     }
730     zkperms.writeToZookeeper(table.getName(), serialized);
731   }
732 
733   public void writeNamespaceToZooKeeper(String namespace,
734       PermissionCache<TablePermission> tablePerms) {
735     byte[] serialized = new byte[0];
736     if (tablePerms != null) {
737       serialized = AccessControlLists.writePermissionsAsBytes(tablePerms.getAllPermissions(), conf);
738     }
739     zkperms.writeToZookeeper(Bytes.toBytes(AccessControlLists.toNamespaceEntry(namespace)),
740         serialized);
741   }
742 
743   public long getMTime() {
744     return mtime;
745   }
746 
747   static Map<ZooKeeperWatcher,TableAuthManager> managerMap =
748     new HashMap<ZooKeeperWatcher,TableAuthManager>();
749 
750   public synchronized static TableAuthManager get(
751       ZooKeeperWatcher watcher, Configuration conf) throws IOException {
752     instance = managerMap.get(watcher);
753     if (instance == null) {
754       instance = new TableAuthManager(watcher, conf);
755       managerMap.put(watcher, instance);
756     }
757     return instance;
758   }
759 }