Metacat

/**

 *  '$RCSfile$'

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

 *              National Center for Ecological Analysis and Synthesis

 *

 *   '$Author: leinfelder $'

 *     '$Date: 2012-03-13 10:43:01 -0700 (Tue, 13 Mar 2012) $'

 *

 * 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.Timer;

import java.util.Vector;

import javax.servlet.http.HttpServletRequest;

import org.apache.commons.io.IOUtils;

import org.apache.log4j.Logger;

import org.dataone.client.CNode;

import org.dataone.client.D1Client;

import org.dataone.client.ObjectFormatCache;

import org.dataone.service.util.Constants;

import org.dataone.service.util.DateTimeMarshaller;

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.AccessPolicy;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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.database.DBConnection;

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

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

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

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

import edu.ucsb.nceas.metacat.util.DocumentUtil;

import edu.ucsb.nceas.metacat.util.SystemUtil;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

public abstract class D1NodeService {

  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;

  /**

   * 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;

	}

  /**

   * 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;

  }

  /**

   * 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 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...");

    // 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.");

    }

    // check that we are not attempting to subvert versioning

    if (sysmeta.getObsoletes() != null && sysmeta.getObsoletes().getValue() != null) {

    	throw new InvalidSystemMetadata("1180", 

    			"The supplied system metadata is invalid. " +

    			"The obsoletes field cannot have a value when creating entries.");

    }

    if (sysmeta.getObsoletedBy() != null && sysmeta.getObsoletedBy().getValue() != null) {

    	throw new InvalidSystemMetadata("1180", 

    			"The supplied system metadata is invalid. " +

    			"The obsoletedBy field cannot have a value when creating entries.");

	}

    // 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()) {

	    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)) {

				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 ) {

      // 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(objectAsXML, 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);

      }   

    }

    // save the sysmeta

    try {

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

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

    } 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;

    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, 

      Event event, Integer start, Integer count) throws InvalidToken, ServiceFailure,

      NotAuthorized, InvalidRequest, NotImplemented {

    Log log = new Log();

    List<LogEntry> logs = new Vector<LogEntry>();

    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;

    }

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

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

    String report = el.getReport(null, null, null, null,

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

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

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

    String logEntry = "<logEntry>";

    String endLogEntry = "</logEntry>";

    int startIndex = 0;

    int foundIndex = report.indexOf(logEntry, startIndex);

    while (foundIndex != -1) {

      // parse out each entry

      int endEntryIndex = report.indexOf(endLogEntry, foundIndex);

      String entry = report.substring(foundIndex, endEntryIndex);

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

      startIndex = endEntryIndex + endLogEntry.length();

      foundIndex = report.indexOf(logEntry, startIndex);

      String entryId = getLogEntryField("entryid", entry);

      String ipAddress = getLogEntryField("ipAddress", entry);

      String principal = getLogEntryField("principal", entry);

      String userAgent = getLogEntryField("userAgent", entry);

      String docid = getLogEntryField("docid", entry);

      String eventS = getLogEntryField("event", entry);

      String dateLogged = getLogEntryField("dateLogged", entry);

      LogEntry le = new LogEntry();

      // handle Metacat -> DataONE event conversion

      if (eventS.equalsIgnoreCase("insert")) {

    	  eventS = Event.CREATE.xmlValue();

      }

      Event e = Event.convert(eventS);

      if (e == null) { // skip any events that are not Dataone Crud events

          logMetacat.warn("skipping unknown event type: '" + eventS + "'");

          continue;

      }

      le.setEvent(e);

      le.setEntryId(entryId);

      Identifier identifier = new Identifier();

      try {

        logMetacat.debug("converting docid '" + docid + "' to a pid.");

        if (docid == null || docid.trim().equals("") || docid.trim().equals("null")) {

          continue;

        }

        String docidNoRev = DocumentUtil.getSmartDocId(docid);

        //docid = docid.substring(0, docid.lastIndexOf("."));

        int rev = DocumentUtil.getRevisionFromAccessionNumber(docid);

        //int rev = im.getLatestRevForLocalId(docid);

        String guid = im.getGUID(docidNoRev, rev);

        identifier.setValue(guid);

      } catch (Exception ex) { 

        // try to get the pid, if that doesn't

        // work, just use the local id

        // throw new ServiceFailure("1030",

        // "Error getting pid for localId " +

        // docid + ": " + ex.getMessage());\

          logMetacat.warn("could not find pid for docid '" + docid + "'");

        // skip it if the pid can't be found

        continue;

      }

      // skip if we are not allowed to read the document in question

      // https://redmine.dataone.org/issues/2444

      boolean allowed = false;

      try {

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

      } catch (NotAuthorized ignore) {}

      catch (NotFound nf) {

    	  logMetacat.warn("Could not check authorization for pid: " + identifier.getValue(), nf);

      }

      if (!allowed) {

    	  logMetacat.debug(Permission.READ + " not allowed on document: " + identifier.getValue());

    	  continue;

      }

      le.setIdentifier(identifier);

      le.setIpAddress(ipAddress);

      Date logDate = DateTimeMarshaller.deserializeDateToUTC(dateLogged);

      le.setDateLogged(logDate);

      NodeReference memberNode = new NodeReference();

      String nodeId = "localhost";

      try {

          nodeId = PropertyService.getProperty("dataone.nodeId");

      } catch (PropertyNotFoundException e1) {

          // TODO Auto-generated catch block

          e1.printStackTrace();

      }

      memberNode.setValue(nodeId);

      le.setNodeIdentifier(memberNode);

      Subject princ = new Subject();

      princ.setValue(principal);

      le.setSubject(princ);

      le.setUserAgent(userAgent);

      // event filtering?

      if (event == null) {

    	  logs.add(le);

      } else if (le.getEvent().equals(event)) {

    	  logs.add(le);

      }

    }

    // d1 paging

    int total = logs.size();

    if (start != null && count != null) {

    	int toIndex = start + count;

    	// do not exceed total

    	toIndex = Math.min(toIndex, total);

    	// do not start greater than total

    	start = Math.min(start, total);

    	// sub set of the list

    	logs = new ArrayList<LogEntry>(logs.subList(start, toIndex));

    }

    log.setLogEntryList(logs);

    log.setStart(start);

    log.setCount(logs.size());

    log.setTotal(total);

    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) {

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

            pid.getValue() +

            "could not be returned due to error: " +

            e.getMessage());

      }

    }

    // 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;

	  try {

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

	  } catch (InvalidRequest e) {

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

	  }

      if (!isAuthorized) {

        throw new NotAuthorized("1400", Permission.READ + " not allowed on " + pid.getValue());  

      }

      SystemMetadata systemMetadata = null;

      try {

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

      } catch (Exception e) {

        // convert hazelcast RuntimeException to NotFound

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

      }

    return systemMetadata;

  }

  /**

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

   * for the operation on the specified object.

   * 

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

   * @param pid - The identifer of the resource for which access is being checked

   * @param operation - The type of operation which is being requested for the given pid

   * 

   * @return true if the operation is allowed

   * 

   * @throws ServiceFailure

   * @throws InvalidToken

   * @throws NotFound

   * @throws NotAuthorized

   * @throws NotImplemented

   */

  protected boolean isAdminAuthorized(Session session, Identifier pid,

      Permission permission) 

      throws ServiceFailure, InvalidToken, NotFound, NotAuthorized,

      NotImplemented {

      boolean allowed = false;

      // are we allowed to do this? only CNs and target MNs are allowed

      CNode cn = D1Client.getCN();

      List<Node> nodes = cn.listNodes().getNodeList();

      if ( nodes == null ) {

          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) {

                  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 authorization 

   * for the operation on the specified object.

   * 

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

   * @param pid - The identifer of the resource for which access is being checked

   * @param operation - The type of operation which is being requested for the given pid

   *

   * @return true if the operation is allowed

   * 

   * @throws ServiceFailure

   * @throws InvalidToken

   * @throws NotFound

   * @throws NotAuthorized

   * @throws NotImplemented

   * @throws InvalidRequest

   */

  public boolean isAuthorized(Session session, Identifier pid, Permission permission)

    throws ServiceFailure, InvalidToken, NotFound, NotAuthorized,

    NotImplemented, InvalidRequest {

    boolean allowed = false;

    if (permission == null) {

    	throw new InvalidRequest("1761", "Permission was not provided or is invalid");

    }

    // permissions are hierarchical

    List<Permission> expandedPermissions = null;

    // for the "Verified" symbolic user

    Subject verifiedSubject = new Subject();

	verifiedSubject.setValue(Constants.SUBJECT_VERIFIED_USER);

    // get the subject[s] from the session

	List<Subject> subjects = new ArrayList<Subject>();

	if (session != null) {

		// primary subject

		Subject subject = session.getSubject();

		if (subject != null) {

			subjects.add(subject);

		}

		// details about the subject

		SubjectInfo subjectInfo = session.getSubjectInfo();

		if (subjectInfo != null) {

			// find subjectInfo for the primary subject

			List<Person> personList = subjectInfo.getPersonList();

			if (personList != null) {

				for (Person p : personList) {

					  // for every person listed (isVerified is transitive)

						logMetacat.debug("checking person");

						logMetacat.debug("p.getVerified(): " + p.getVerified());

						if (p.getVerified() != null && p.getVerified()) {

							// add the verified symbolic user

							if (!subjects.contains(verifiedSubject)) {

								subjects.add(verifiedSubject);

							}

						}

						// add the equivalent identities

						List<Subject> equivList = p

								.getEquivalentIdentityList();

						if (equivList != null) {

							for (Subject equiv : equivList) {

								subjects.add(equiv);

							}

						}

						// add the groups

						List<Subject> groupList = p.getIsMemberOfList();

						if (groupList != null) {

							for (Subject g : groupList) {

								subjects.add(g);

							}

						}

						break;

				}

			}

		}

		// add the authenticated symbolic since we have a session

		Subject authenticatedSubject = new Subject();

		authenticatedSubject.setValue(Constants.SUBJECT_AUTHENTICATED_USER);

		subjects.add(authenticatedSubject);

	}

    // add public subject for everyone

    Subject publicSubject = new Subject();

    publicSubject.setValue(Constants.SUBJECT_PUBLIC);

    subjects.add(publicSubject);

    // get the system metadata

    String pidStr = pid.getValue();

    SystemMetadata systemMetadata = null;

    try {

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

    } catch (Exception e) {

        // convert Hazelcast RuntimeException to NotFound

        logMetacat.error("An error occurred while getting system metadata for identifier " +

            pid.getValue() + ". The error message was: " + e.getMessage());

        throw new NotFound("1800", "No record found for " + pidStr);

    } 

    // throw not found if it was not found

    if (systemMetadata == null) {

    	throw new NotFound("1800", "No system metadata could be found for given PID: " + pidStr);

    }

    // do we own it?

    for (Subject s: subjects) {

      logMetacat.debug("Comparing \t" + 

                       systemMetadata.getRightsHolder().getValue() +

                       " \tagainst \t" + s.getValue());

    	allowed = systemMetadata.getRightsHolder().equals(s);

    	if (allowed) {

    		return allowed;

    	}

    }    

    // otherwise check the access rules

    try {

	    List<AccessRule> allows = systemMetadata.getAccessPolicy().getAllowList();

	    search: // label break

	    for (AccessRule accessRule: allows) {

	      for (Subject s: subjects) {

	        logMetacat.debug("Checking allow access rule for subject: " + s.getValue());

	        if (accessRule.getSubjectList().contains(s)) {

	        	logMetacat.debug("Access rule contains subject: " + s.getValue());

	        	for (Permission p: accessRule.getPermissionList()) {

		        	logMetacat.debug("Checking permission: " + p.xmlValue());

	        		expandedPermissions = expandPermissions(p);

	        		allowed = expandedPermissions.contains(permission);

	        		if (allowed) {

			        	logMetacat.info("Permission granted: " + p.xmlValue() + " to " + s.getValue());

	        			break search; //label break

	        		}

	        	}

	        }

	      }

	    }

    } catch (Exception e) {

    	// catch all for errors - safe side should be to deny the access

    	logMetacat.error("Problem checking authorization - defaulting to deny", e);

		allowed = false;

    }

    // throw or return?

    if (!allowed) {

      throw new NotAuthorized("1820", permission + " not allowed on " + pidStr);

    }

    return allowed;

  }

  /*

   * parse a logEntry and get the relevant field from it

   * 

   * @param fieldname

   * @param entry

   * @return

   */

  private String getLogEntryField(String fieldname, String entry) {

    String begin = "<" + fieldname + ">";

    String end = "</" + fieldname + ">";

    // logMetacat.debug("looking for " + begin + " and " + end +

    // " in entry " + entry);

    String s = entry.substring(entry.indexOf(begin) + begin.length(), entry

        .indexOf(end));

    logMetacat.debug("entry " + fieldname + " : " + s);

    return s;

  }

  /** 

   * Determine if a given object should be treated as an XML science metadata

   * object. 

   * 

   * @param sysmeta - the SystemMetadata describing the object

   * @return true if the object should be treated as science metadata

   */

  public static boolean isScienceMetadata(SystemMetadata sysmeta) {

    ObjectFormat objectFormat = null;

    boolean isScienceMetadata = false;

    try {

      objectFormat = ObjectFormatCache.getInstance().getFormat(sysmeta.getFormatId());

      if ( objectFormat.getFormatType().equals("METADATA") ) {

      	isScienceMetadata = true;

      }

    } catch (ServiceFailure e) {

      logMetacat.debug("There was a problem determining if the object identified by" + 

          sysmeta.getIdentifier().getValue() + 

          " is science metadata: " + e.getMessage());

    } catch (NotFound e) {

      logMetacat.debug("There was a problem determining if the object identified by" + 

          sysmeta.getIdentifier().getValue() + 

          " is science metadata: " + e.getMessage());

    }

    return isScienceMetadata;

  }

  /**

   * Insert or update an XML document into Metacat

   * 

   * @param xml - the XML document to insert or update

   * @param pid - the identifier to be used for the resulting object

   * 

   * @return localId - the resulting docid of the document created or updated

   *