Metacat

/**

 *  '$RCSfile$'

 *  Copyright: 2000-2011 Regents of the University of California and the

 *              National Center for Ecological Analysis and Synthesis

 *

 *   '$Author: tao $'

 *     '$Date: 2014-10-31 17:36:28 -0700 (Fri, 31 Oct 2014) $'

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; either version 2 of the License, or

 * (at your option) any later version.

 *

 * This program is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 * GNU General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

 */

package edu.ucsb.nceas.metacat.dataone;

import java.io.File;

import java.io.FileNotFoundException;

import java.io.FileOutputStream;

import java.io.IOException;

import java.io.InputStream;

import java.io.OutputStream;

import java.sql.SQLException;

import java.util.ArrayList;

import java.util.Calendar;

import java.util.Date;

import java.util.Hashtable;

import java.util.List;

import java.util.Set;

import java.util.Timer;

import java.util.concurrent.locks.Lock;

import javax.servlet.http.HttpServletRequest;

import org.apache.commons.io.IOUtils;

import org.apache.log4j.Logger;

import org.dataone.client.v2.CNode;

import org.dataone.client.v2.itk.D1Client;

import org.dataone.client.v2.formats.ObjectFormatCache;

import org.dataone.service.exceptions.BaseException;

import org.dataone.service.exceptions.IdentifierNotUnique;

import org.dataone.service.exceptions.InsufficientResources;

import org.dataone.service.exceptions.InvalidRequest;

import org.dataone.service.exceptions.InvalidSystemMetadata;

import org.dataone.service.exceptions.InvalidToken;

import org.dataone.service.exceptions.NotAuthorized;

import org.dataone.service.exceptions.NotFound;

import org.dataone.service.exceptions.NotImplemented;

import org.dataone.service.exceptions.ServiceFailure;

import org.dataone.service.exceptions.UnsupportedType;

import org.dataone.service.types.v1.AccessRule;

import org.dataone.service.types.v1.DescribeResponse;

import org.dataone.service.types.v1.Group;

import org.dataone.service.types.v1.Identifier;

import org.dataone.service.types.v2.Log;

import org.dataone.service.types.v2.Node;

import org.dataone.service.types.v1.Event;

import org.dataone.service.types.v1.NodeReference;

import org.dataone.service.types.v1.NodeType;

import org.dataone.service.types.v2.ObjectFormat;

import org.dataone.service.types.v1.Permission;

import org.dataone.service.types.v1.Replica;

import org.dataone.service.types.v1.Session;

import org.dataone.service.types.v1.Subject;

import org.dataone.service.types.v2.SystemMetadata;

import org.dataone.service.types.v1.util.AuthUtils;

import org.dataone.service.types.v1.util.ChecksumUtil;

import org.dataone.service.util.Constants;

import edu.ucsb.nceas.metacat.AccessionNumberException;

import edu.ucsb.nceas.metacat.DocumentImpl;

import edu.ucsb.nceas.metacat.EventLog;

import edu.ucsb.nceas.metacat.IdentifierManager;

import edu.ucsb.nceas.metacat.McdbDocNotFoundException;

import edu.ucsb.nceas.metacat.MetacatHandler;

import edu.ucsb.nceas.metacat.client.InsufficientKarmaException;

import edu.ucsb.nceas.metacat.database.DBConnection;

import edu.ucsb.nceas.metacat.database.DBConnectionPool;

import edu.ucsb.nceas.metacat.dataone.hazelcast.HazelcastService;

import edu.ucsb.nceas.metacat.index.MetacatSolrIndex;

import edu.ucsb.nceas.metacat.properties.PropertyService;

import edu.ucsb.nceas.metacat.replication.ForceReplicationHandler;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

public abstract class D1NodeService {

  public static final String DELETEDMESSAGE = "The object with the PID has been deleted from the node.";

  private static Logger logMetacat = Logger.getLogger(D1NodeService.class);

  /** For logging the operations */

  protected HttpServletRequest request;

  /* reference to the metacat handler */

  protected MetacatHandler handler;

  /* parameters set in the incoming request */

  private Hashtable<String, String[]> params;

  /**

   * limit paged results sets to a configured maximum

   */

  protected static int MAXIMUM_DB_RECORD_COUNT = 7000;

  static {

		try {

			MAXIMUM_DB_RECORD_COUNT = Integer.valueOf(PropertyService.getProperty("database.webResultsetSize"));

		} catch (Exception e) {

			logMetacat.warn("Could not set MAXIMUM_DB_RECORD_COUNT", e);

		}

	}

  /**

   * out-of-band session object to be used when not passed in as a method parameter

   */

  protected Session session;

  /**

   * Constructor - used to set the metacatUrl from a subclass extending D1NodeService

   * 

   * @param metacatUrl - the URL of the metacat service, including the ending /d1

   */

  public D1NodeService(HttpServletRequest request) {

		this.request = request;

	}

  /**

   * retrieve the out-of-band session

   * @return

   */

  	public Session getSession() {

		return session;

	}

  	/**

  	 * Set the out-of-band session

  	 * @param session

  	 */

	public void setSession(Session session) {

		this.session = session;

	}

  /**

   * This method provides a lighter weight mechanism than 

   * getSystemMetadata() for a client to determine basic 

   * properties of the referenced object.

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the identifier of the object to be described

   * 

   * @return describeResponse - A set of values providing a basic description 

   *                            of the object.

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   * @throws InvalidRequest

   */

  public DescribeResponse describe(Session session, Identifier pid) 

      throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, NotImplemented {

    // get system metadata and construct the describe response

      SystemMetadata sysmeta = getSystemMetadata(session, pid);

      DescribeResponse describeResponse = 

      	new DescribeResponse(sysmeta.getFormatId(), sysmeta.getSize(), 

      			sysmeta.getDateSysMetadataModified(),

      			sysmeta.getChecksum(), sysmeta.getSerialVersion());

      return describeResponse;

  }

  /**

   * Deletes an object from the Member Node, where the object is either a 

   * data object or a science metadata object.

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - The object identifier to be deleted

   * 

   * @return pid - the identifier of the object used for the deletion

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   * @throws InvalidRequest

   */

  public Identifier delete(Session session, Identifier pid) 

      throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, NotImplemented {

      String localId = null;

      if (session == null) {

      	throw new InvalidToken("1330", "No session has been provided");

      }

      // just for logging purposes

      String username = session.getSubject().getValue();

      // do we have a valid pid?

      if (pid == null || pid.getValue().trim().equals("")) {

          throw new ServiceFailure("1350", "The provided identifier was invalid.");

      }

      // check for the existing identifier

      try {

          localId = IdentifierManager.getInstance().getLocalId(pid.getValue());

      } catch (McdbDocNotFoundException e) {

          throw new NotFound("1340", "The object with the provided " + "identifier was not found.");

      }

      try {

          // delete the document, as admin

          DocumentImpl.delete(localId, null, null, null, true);

          EventLog.getInstance().log(request.getRemoteAddr(), request.getHeader("User-Agent"), username, localId, Event.DELETE.xmlValue());

          // archive it

          // DocumentImpl.delete() now sets this

          // see https://redmine.dataone.org/issues/3406

//          SystemMetadata sysMeta = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

//          sysMeta.setArchived(true);

//          sysMeta.setDateSysMetadataModified(Calendar.getInstance().getTime());

//          HazelcastService.getInstance().getSystemMetadataMap().put(pid, sysMeta);

      } catch (McdbDocNotFoundException e) {

          throw new NotFound("1340", "The provided identifier was invalid.");

      } catch (SQLException e) {

          throw new ServiceFailure("1350", "There was a problem deleting the object." + "The error message was: " + e.getMessage());

      } catch (InsufficientKarmaException e) {

          if ( logMetacat.isDebugEnabled() ) {

              e.printStackTrace();

          }

          throw new NotAuthorized("1320", "The provided identity does not have " + "permission to DELETE objects on the Member Node.");

      } catch (Exception e) { // for some reason DocumentImpl throws a general Exception

          throw new ServiceFailure("1350", "There was a problem deleting the object." + "The error message was: " + e.getMessage());

      }

      return pid;

  }

  /**

   * Low level, "are you alive" operation. A valid ping response is 

   * indicated by a HTTP status of 200.

   * 

   * @return true if the service is alive

   * 

   * @throws NotImplemented

   * @throws ServiceFailure

   * @throws InsufficientResources

   */

  public Date ping() 

      throws NotImplemented, ServiceFailure, InsufficientResources {

      // test if we can get a database connection

      int serialNumber = -1;

      DBConnection dbConn = null;

      try {

          dbConn = DBConnectionPool.getDBConnection("MNodeService.ping");

          serialNumber = dbConn.getCheckOutSerialNumber();

      } catch (SQLException e) {

      	ServiceFailure sf = new ServiceFailure("", e.getMessage());

      	sf.initCause(e);

          throw sf;

      } finally {

          // Return the database connection

          DBConnectionPool.returnDBConnection(dbConn, serialNumber);

      }

      return Calendar.getInstance().getTime();

  }

  /**

   * Adds a new object to the Node, where the object is either a data 

   * object or a science metadata object. This method is called by clients 

   * to create new data objects on Member Nodes or internally for Coordinating

   * Nodes

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - The object identifier to be created

   * @param object - the object bytes

   * @param sysmeta - the system metadata that describes the object  

   * 

   * @return pid - the object identifier created

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws IdentifierNotUnique

   * @throws UnsupportedType

   * @throws InsufficientResources

   * @throws InvalidSystemMetadata

   * @throws NotImplemented

   * @throws InvalidRequest

   */

  public Identifier create(Session session, Identifier pid, InputStream object,

    SystemMetadata sysmeta) 

    throws InvalidToken, ServiceFailure, NotAuthorized, IdentifierNotUnique, 

    UnsupportedType, InsufficientResources, InvalidSystemMetadata, 

    NotImplemented, InvalidRequest {

    Identifier resultPid = null;

    String localId = null;

    boolean allowed = false;

    // check for null session

    if (session == null) {

    	throw new InvalidToken("4894", "Session is required to WRITE to the Node.");

    }

    Subject subject = session.getSubject();

    Subject publicSubject = new Subject();

    publicSubject.setValue(Constants.SUBJECT_PUBLIC);

	// be sure the user is authenticated for create()

    if (subject == null || subject.getValue() == null || 

        subject.equals(publicSubject) ) {

      throw new NotAuthorized("1100", "The provided identity does not have " +

        "permission to WRITE to the Node.");

    }

    // verify the pid is valid format

    if (!isValidIdentifier(pid)) {

    	throw new InvalidRequest("1202", "The provided identifier is invalid.");

    }

    // verify that pid == SystemMetadata.getIdentifier()

    logMetacat.debug("Comparing pid|sysmeta_pid: " + 

      pid.getValue() + "|" + sysmeta.getIdentifier().getValue());

    if (!pid.getValue().equals(sysmeta.getIdentifier().getValue())) {

        throw new InvalidSystemMetadata("1180", 

            "The supplied system metadata is invalid. " +

            "The identifier " + pid.getValue() + " does not match identifier" +

            "in the system metadata identified by " +

            sysmeta.getIdentifier().getValue() + ".");

    }

    logMetacat.debug("Checking if identifier exists: " + pid.getValue());

    // Check that the identifier does not already exist

    if (IdentifierManager.getInstance().identifierExists(pid.getValue())) {

	    	throw new IdentifierNotUnique("1120", 

			          "The requested identifier " + pid.getValue() +

			          " is already used by another object and" +

			          "therefore can not be used for this object. Clients should choose" +

			          "a new identifier that is unique and retry the operation or " +

			          "use CN.reserveIdentifier() to reserve one.");

    }

    // TODO: this probably needs to be refined more

    try {

      allowed = isAuthorized(session, pid, Permission.WRITE);

    } catch (NotFound e) {

      // The identifier doesn't exist, writing should be fine.

      allowed = true;

    }

    // verify checksum, only if we can reset the inputstream

    if (object.markSupported()) {

        logMetacat.debug("Checking checksum for: " + pid.getValue());

	    String checksumAlgorithm = sysmeta.getChecksum().getAlgorithm();

	    String checksumValue = sysmeta.getChecksum().getValue();

	    try {

			String computedChecksumValue = ChecksumUtil.checksum(object, checksumAlgorithm).getValue();

			// it's very important that we don't consume the stream

			object.reset();

			if (!computedChecksumValue.equals(checksumValue)) {

			    logMetacat.error("Checksum for " + pid.getValue() + " does not match system metadata, computed = " + computedChecksumValue );

				throw new InvalidSystemMetadata("4896", "Checksum given does not match that of the object");

			}

		} catch (Exception e) {

			String msg = "Error verifying checksum values";

	      	logMetacat.error(msg, e);

	        throw new ServiceFailure("1190", msg + ": " + e.getMessage());

		}

    } else {

    	logMetacat.warn("mark is not supported on the object's input stream - cannot verify checksum without consuming stream");

    }

    // we have the go ahead

    if ( allowed ) {

        logMetacat.debug("Allowed to insert: " + pid.getValue());

      // Science metadata (XML) or science data object?

      // TODO: there are cases where certain object formats are science metadata

      // but are not XML (netCDF ...).  Handle this.

      if ( isScienceMetadata(sysmeta) ) {

        // CASE METADATA:

      	//String objectAsXML = "";

        try {

	        //objectAsXML = IOUtils.toString(object, "UTF-8");

	        localId = insertOrUpdateDocument(object,"UTF-8", pid, session, "insert");

	        //localId = im.getLocalId(pid.getValue());

        } catch (IOException e) {

        	String msg = "The Node is unable to create the object. " +

          "There was a problem converting the object to XML";

        	logMetacat.info(msg);

          throw new ServiceFailure("1190", msg + ": " + e.getMessage());

        }

      } else {

	      // DEFAULT CASE: DATA (needs to be checked and completed)

	      localId = insertDataObject(object, pid, session);

      }   

    }

    logMetacat.debug("Done inserting new object: " + pid.getValue());

    // save the sysmeta

    try {

    	// lock and unlock of the pid happens in the subclass

    	HazelcastService.getInstance().getSystemMetadataMap().put(sysmeta.getIdentifier(), sysmeta);

    	// submit for indexing

        MetacatSolrIndex.getInstance().submit(sysmeta.getIdentifier(), sysmeta, null, true);

    } catch (Exception e) {

    	logMetacat.error("Problem creating system metadata: " + pid.getValue(), e);

        throw new ServiceFailure("1190", e.getMessage());

	}

    // setting the resulting identifier failed

    if (localId == null ) {

      throw new ServiceFailure("1190", "The Node is unable to create the object. ");

    }

    resultPid = pid;

    logMetacat.debug("create() complete for object: " + pid.getValue());

    return resultPid;

  }

  /**

   * Return the log records associated with a given event between the start and 

   * end dates listed given a particular Subject listed in the Session

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param fromDate - the start date of the desired log records

   * @param toDate - the end date of the desired log records

   * @param event - restrict log records of a specific event type

   * @param start - zero based offset from the first record in the 

   *                set of matching log records. Used to assist with 

   *                paging the response.

   * @param count - maximum number of log records to return in the response. 

   *                Used to assist with paging the response.

   * 

   * @return the desired log records

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  public Log getLogRecords(Session session, Date fromDate, Date toDate, 

      String event, String pidFilter, Integer start, Integer count) throws InvalidToken, ServiceFailure,

      NotAuthorized, InvalidRequest, NotImplemented {

	  // only admin access to this method

	  // see https://redmine.dataone.org/issues/2855

	  if (!isAdminAuthorized(session)) {

		  throw new NotAuthorized("1460", "Only the CN or admin is allowed to harvest logs from this node");

	  }

    IdentifierManager im = IdentifierManager.getInstance();

    EventLog el = EventLog.getInstance();

    if ( fromDate == null ) {

      logMetacat.debug("setting fromdate from null");

      fromDate = new Date(1);

    }

    if ( toDate == null ) {

      logMetacat.debug("setting todate from null");

      toDate = new Date();

    }

    if ( start == null ) {

    	start = 0;	

    }

    if ( count == null ) {

    	count = 1000;

    }

    // safeguard against large requests

    if (count > MAXIMUM_DB_RECORD_COUNT) {

    	count = MAXIMUM_DB_RECORD_COUNT;

    }

    String[] filterDocid = null;

    if (pidFilter != null) {

		try {

	      String localId = im.getLocalId(pidFilter);

	      filterDocid = new String[] {localId};

	    } catch (Exception ex) { 

	    	String msg = "Could not find localId for given pidFilter '" + pidFilter + "'";

	        logMetacat.warn(msg, ex);

	        //throw new InvalidRequest("1480", msg);

	    }

    }

    logMetacat.debug("fromDate: " + fromDate);

    logMetacat.debug("toDate: " + toDate);

    Log log = el.getD1Report(null, null, filterDocid, event,

        new java.sql.Timestamp(fromDate.getTime()),

        new java.sql.Timestamp(toDate.getTime()), false, start, count);

    logMetacat.info("getLogRecords");

    return log;

  }

  /**

   * Return the object identified by the given object identifier

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * 

   * TODO: The D1 Authorization API doesn't provide information on which 

   * authentication system the Subject belongs to, and so it's not possible to

   * discern which Person or Group is a valid KNB LDAP DN.  Fix this.

   * 

   * @return inputStream - the input stream of the given object

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  public InputStream get(Session session, Identifier pid) 

    throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, 

    NotImplemented {

    InputStream inputStream = null; // bytes to be returned

    handler = new MetacatHandler(new Timer());

    boolean allowed = false;

    String localId; // the metacat docid for the pid

    // get the local docid from Metacat

    try {

      localId = IdentifierManager.getInstance().getLocalId(pid.getValue());

    } catch (McdbDocNotFoundException e) {

      throw new NotFound("1020", "The object specified by " + 

                         pid.getValue() +

                         " does not exist at this node.");

    }

    // check for authorization

    try {

		allowed = isAuthorized(session, pid, Permission.READ);

	} catch (InvalidRequest e) {

		throw new ServiceFailure("1030", e.getDescription());

	}

    // if the person is authorized, perform the read

    if (allowed) {

      try {

        inputStream = handler.read(localId);

      } catch (Exception e) {

        String error ="";

        if(EventLog.getInstance().isDeleted(localId)) {

              error=DELETEDMESSAGE;

        }

        throw new NotFound("1020", "The object specified by " + 

            pid.getValue() +

            "could not be returned due to error: " +

            e.getMessage()+". "+error);

      }

    }

    // if we fail to set the input stream

    if ( inputStream == null ) {

      throw new NotFound("1020", "The object specified by " + 

                         pid.getValue() +

                         "does not exist at this node.");

    }

	// log the read event

    String principal = Constants.SUBJECT_PUBLIC;

    if (session != null && session.getSubject() != null) {

    	principal = session.getSubject().getValue();

    }

    EventLog.getInstance().log(request.getRemoteAddr(), request.getHeader("User-Agent"), principal, localId, "read");

    return inputStream;

  }

  /**

   * Return the system metadata for a given object

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * 

   * @return inputStream - the input stream of the given system metadata object

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws InvalidRequest

   * @throws NotImplemented

   */

    public SystemMetadata getSystemMetadata(Session session, Identifier pid)

        throws InvalidToken, ServiceFailure, NotAuthorized, NotFound,

        NotImplemented {

        boolean isAuthorized = false;

        SystemMetadata systemMetadata = null;

        List<Replica> replicaList = null;

        NodeReference replicaNodeRef = null;

        List<Node> nodeListBySubject = null;

        Subject subject = null;

        if (session != null ) {

            subject = session.getSubject();

        }

        // check normal authorization

        BaseException originalAuthorizationException = null;

        if (!isAuthorized) {

            try {

                isAuthorized = isAuthorized(session, pid, Permission.READ);

            } catch (InvalidRequest e) {

                throw new ServiceFailure("1090", e.getDescription());

            } catch (NotAuthorized nae) {

            	// catch this for later

            	originalAuthorizationException = nae;

			}

        }

        // get the system metadata first because we need the replica list for auth

        systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

        // check the replica information to expand access to MNs that might need it

        if (!isAuthorized) {

	        try {

	            // if MNs are listed as replicas, allow access

	            if ( systemMetadata != null ) {

	                replicaList = systemMetadata.getReplicaList();

	                // only check if there are in fact replicas listed

	                if ( replicaList != null ) {

	                    if ( subject != null ) {

	                        // get the list of nodes with a matching node subject

	                        try {

	                            nodeListBySubject = listNodesBySubject(session.getSubject());

	                        } catch (BaseException e) {

	                            // Unexpected error contacting the CN via D1Client

	                            String msg = "Caught an unexpected error while trying "

	                                    + "to potentially authorize system metadata access "

	                                    + "based on the session subject. The error was "

	                                    + e.getMessage();

	                            logMetacat.error(msg);

	                            if (logMetacat.isDebugEnabled()) {

	                                e.printStackTrace();

	                            }

	                            // isAuthorized is still false 

	                        }

	                    }

	                    if (nodeListBySubject != null) {

	                        // compare node ids to replica node ids

	                        outer: for (Replica replica : replicaList) {

	                            replicaNodeRef = replica.getReplicaMemberNode();

	                            for (Node node : nodeListBySubject) {

	                                if (node.getIdentifier().equals(replicaNodeRef)) {

	                                    // node id via session subject matches a replica node

	                                    isAuthorized = true;

	                                    break outer;

	                                }

	                            }

	                        }

	                    }

	                }

	            }

	            // if we still aren't authorized, then we are done

	            if (!isAuthorized) {

	                throw new NotAuthorized("1400", Permission.READ

	                        + " not allowed on " + pid.getValue());

	            }

	        } catch (RuntimeException e) {

	        	e.printStackTrace();

	            // convert hazelcast RuntimeException to ServiceFailure

	            throw new ServiceFailure("1090", "Unexpected error getting system metadata for: " + 

	                pid.getValue());	

	        }

        }

        // It wasn't in the map

        if ( systemMetadata == null ) {

            String error ="";

            String localId = null;

            try {

                localId = IdentifierManager.getInstance().getLocalId(pid.getValue());

             } catch (Exception e) {

                logMetacat.warn("Couldn't find the local id for the pid "+pid.getValue());

            }

            if(localId != null && EventLog.getInstance().isDeleted(localId)) {

                error = DELETEDMESSAGE;

            }

            throw new NotFound("1420", "No record found for: " + pid.getValue()+". "+error);

        }

        return systemMetadata;

    }

    /**

     * Test if the specified session represents the authoritative member node for the

     * given object specified by the identifier. According the the DataONE documentation, 

     * the authoritative member node has all the rights of the *rightsHolder*.

     * @param session - the Session object containing the credentials for the Subject

     * @param pid - the Identifier of the data object

     * @return true if the session represents the authoritative mn.

     * @throws ServiceFailure 

     * @throws NotImplemented 

     */

    public boolean isAuthoritativeMNodeAdmin(Session session, Identifier pid) {

        boolean allowed = false;

        //check the parameters

        if(session == null) {

            logMetacat.debug("D1NodeService.isAuthoritativeMNodeAdmin - the session object is null and return false.");

            return allowed;

        } else if (pid == null || pid.getValue() == null || pid.getValue().trim().equals("")) {

            logMetacat.debug("D1NodeService.isAuthoritativeMNodeAdmin - the Identifier object is null (not being specified) and return false.");

            return allowed;

        }

        //Get the subject from the session

        Subject subject = session.getSubject();

        if(subject != null) {

            //Get the authoritative member node info from the system metadata

            SystemMetadata sysMeta = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

            if(sysMeta != null) {

                NodeReference authoritativeMNode = sysMeta.getAuthoritativeMemberNode();

                if(authoritativeMNode != null) {

                        CNode cn = null;

                        try {

                            cn = D1Client.getCN();

                        } catch (BaseException e) {

                            logMetacat.error("D1NodeService.isAuthoritativeMNodeAdmin - couldn't connect to the CN since "+

                                            e.getDescription()+ ". The false value will be returned for the AuthoritativeMNodeAdmin.");

                            return allowed;

                        }

                        if(cn != null) {

                            List<Node> nodes = null;

                            try {

                                nodes = cn.listNodes().getNodeList();

                            } catch (NotImplemented e) {

                                logMetacat.error("D1NodeService.isAuthoritativeMNodeAdmin - couldn't get the member nodes list from the CN since "+e.getDescription()+ 

                                                ". The false value will be returned for the AuthoritativeMNodeAdmin.");

                                return allowed;

                            } catch (ServiceFailure ee) {

                                logMetacat.error("D1NodeService.isAuthoritativeMNodeAdmin - couldn't get the member nodes list from the CN since "+ee.getDescription()+ 

                                                ". The false value will be returned for the AuthoritativeMNodeAdmin.");

                                return allowed;

                            }

                            if(nodes != null) {

                                for(Node node : nodes) {

                                    //find the authoritative node and get its subjects

                                    if (node.getType() == NodeType.MN && node.getIdentifier() != null && node.getIdentifier().equals(authoritativeMNode)) {

                                        List<Subject> nodeSubjects = node.getSubjectList();

                                        if(nodeSubjects != null) {

                                            // check if the session subject is in the node subject list

                                            for (Subject nodeSubject : nodeSubjects) {

                                                logMetacat.debug("D1NodeService.isAuthoritativeMNodeAdmin(), comparing subjects: " +

                                                    nodeSubject.getValue() + " and " + subject.getValue());

                                                if ( nodeSubject != null && nodeSubject.equals(subject) ) {

                                                    allowed = true; // subject of session == target node subject

                                                    break;

                                                }

                                            }              

                                        }

                                    }

                                }

                            }

                        }

                }

            }

        }

        return allowed;

    }

  /**

   * Test if the user identified by the provided token has administrative authorization 

   * 

   * @param session - the Session object containing the credentials for the Subject

   * 

   * @return true if the user is admin

   * 

   * @throws ServiceFailure

   * @throws InvalidToken

   * @throws NotFound

   * @throws NotAuthorized

   * @throws NotImplemented

   */

  public boolean isAdminAuthorized(Session session) 

      throws ServiceFailure, InvalidToken, NotAuthorized,

      NotImplemented {

      boolean allowed = false;

      // must have a session in order to check admin 

      if (session == null) {

         logMetacat.debug("In isAdminAuthorized(), session is null ");

         return false;

      }

      logMetacat.debug("In isAdminAuthorized(), checking CN or MN authorization for " +

           session.getSubject().getValue());

      // check if this is the node calling itself (MN)

      allowed = isNodeAdmin(session);

      // check the CN list

      if (!allowed) {

	      List<Node> nodes = null;

    	  try {

		      // are we allowed to do this? only CNs are allowed

		      CNode cn = D1Client.getCN();

		      nodes = cn.listNodes().getNodeList();

    	  }

	      catch (Throwable e) {

	    	  logMetacat.warn(e.getMessage());

	    	  return false;  

	      }

	      if ( nodes == null ) {

	    	  return false;

	          //throw new ServiceFailure("4852", "Couldn't get node list.");

	      }

	      // find the node in the node list

	      for ( Node node : nodes ) {

	          NodeReference nodeReference = node.getIdentifier();

	          logMetacat.debug("In isAdminAuthorized(), Node reference is: " + nodeReference.getValue());

	          Subject subject = session.getSubject();

	          if (node.getType() == NodeType.CN) {

	              List<Subject> nodeSubjects = node.getSubjectList();

	              // check if the session subject is in the node subject list

	              for (Subject nodeSubject : nodeSubjects) {

	                  logMetacat.debug("In isAdminAuthorized(), comparing subjects: " +

	                      nodeSubject.getValue() + " and " + subject.getValue());

	                  if ( nodeSubject.equals(subject) ) {

	                      allowed = true; // subject of session == target node subject

	                      break;

	                  }

	              }              

	          }

	      }

      }

      return allowed;

  }

  /**

   * Test if the user identified by the provided token has administrative authorization 

   * on this node because they are calling themselves

   * 

   * @param session - the Session object containing the credentials for the Subject

   * 

   * @return true if the user is this node

   * @throws ServiceFailure 

   * @throws NotImplemented 

   */

  public boolean isNodeAdmin(Session session) throws NotImplemented, ServiceFailure {

      boolean allowed = false;

      // must have a session in order to check admin 

      if (session == null) {

         logMetacat.debug("In isNodeAdmin(), session is null ");

         return false;

      }

      logMetacat.debug("In isNodeAdmin(), MN authorization for " +

           session.getSubject().getValue());

      Node node = MNodeService.getInstance(request).getCapabilities();

      NodeReference nodeReference = node.getIdentifier();

      logMetacat.debug("In isNodeAdmin(), Node reference is: " + nodeReference.getValue());

      Subject subject = session.getSubject();

      if (node.getType() == NodeType.MN) {

          List<Subject> nodeSubjects = node.getSubjectList();

          // check if the session subject is in the node subject list

          for (Subject nodeSubject : nodeSubjects) {

              logMetacat.debug("In isNodeAdmin(), comparing subjects: " +

                  nodeSubject.getValue() + " and " + subject.getValue());

              if ( nodeSubject.equals(subject) ) {