commit 7b29aec3183da862b401765cd147014fad487b2a

Author: Vinod Kumar <vinodkv@yahoo-inc.com>
Date:   Sat Feb 27 16:41:17 2010 +0530

    MAPREDUCE-1493 from https://issues.apache.org/jira/secure/attachment/12437336/MAPREDUCE-1493-20100227.3-ydist.txt


git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20-security-patches@1077248 13f79535-47bb-0310-9956-ffa450edef68

src/mapred/org/apache/hadoop/mapred/JSPUtil.java

@@ -31,32 +31,30 @@ import javax.servlet.RequestDispatcher;
 import javax.servlet.ServletException;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
+import javax.servlet.jsp.JspWriter;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.http.HtmlQuoting;
 import org.apache.hadoop.mapred.JobHistory.JobInfo;
+import org.apache.hadoop.mapred.JobHistory.Keys;
 import org.apache.hadoop.mapred.JobTracker.RetireJobInfo;
 import org.apache.hadoop.mapreduce.JobACL;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authorize.AccessControlList;
 import org.apache.hadoop.util.ServletUtil;
 import org.apache.hadoop.util.StringUtils;
 
 class JSPUtil {
   static final String PRIVATE_ACTIONS_KEY = "webinterface.private.actions";
-  
-  public static final Configuration conf = new Configuration();
 
   //LRU based cache
   private static final Map<String, JobInfo> jobHistoryCache = 
     new LinkedHashMap<String, JobInfo>(); 
 
-  private static final int CACHE_SIZE = 
-    conf.getInt("mapred.job.tracker.jobhistory.lru.cache.size", 5);
-
   private static final Log LOG = LogFactory.getLog(JSPUtil.class);
 
   /**
@@ -403,8 +401,7 @@ class JSPUtil {
             "<td id=\"job_" + rowId + "\">" + 
             
               (historyFileUrl == null ? "" :
-              "<a href=\"jobdetailshistory.jsp?jobid=" + 
-              info.status.getJobId() + "&logFile=" + historyFileUrl + "\">") + 
+              "<a href=\"jobdetailshistory.jsp?logFile=" + historyFileUrl + "\">") + 
               
               info.status.getJobId() + "</a></td>" +
             
@@ -438,19 +435,42 @@ class JSPUtil {
     return sb.toString();
   }
 
-  static JobInfo getJobInfo(HttpServletRequest request, FileSystem fs) 
-      throws IOException {
-    String jobid = request.getParameter("jobid");
-    String logFile = request.getParameter("logFile");
+  static Path getJobConfFilePath(Path logFile) {
+    String[] jobDetails = logFile.getName().split("_");
+    String jobId = getJobID(logFile.getName());
+    String jobUniqueString =
+        jobDetails[0] + "_" + jobDetails[1] + "_" + jobId;
+    Path logDir = logFile.getParent();
+    Path jobFilePath = new Path(logDir, jobUniqueString + "_conf.xml");
+    return jobFilePath;
+  }
+
+  /**
+   * Read a job-history log file and construct the corresponding {@link JobInfo}
+   * . Also cache the {@link JobInfo} for quick serving further requests.
+   * 
+   * @param logFile
+   * @param fs
+   * @param jobTracker
+   * @return JobInfo
+   * @throws IOException
+   */
+  static JobInfo getJobInfo(Path logFile, FileSystem fs,
+      JobTracker jobTracker) throws IOException {
+    String jobid = getJobID(logFile.getName());
+    JobInfo jobInfo = null;
     synchronized(jobHistoryCache) {
-      JobInfo jobInfo = jobHistoryCache.remove(jobid);
+      jobInfo = jobHistoryCache.remove(jobid);
       if (jobInfo == null) {
         jobInfo = new JobHistory.JobInfo(jobid);
         LOG.info("Loading Job History file "+jobid + ".   Cache size is " +
             jobHistoryCache.size());
-        DefaultJobHistoryParser.parseJobTasks( logFile, jobInfo, fs) ; 
+        DefaultJobHistoryParser.parseJobTasks(logFile.toUri().getPath(),
+            jobInfo, logFile.getFileSystem(jobTracker.conf));
       }
       jobHistoryCache.put(jobid, jobInfo);
+      int CACHE_SIZE = 
+        jobTracker.conf.getInt("mapred.job.tracker.jobhistory.lru.cache.size", 5);
       if (jobHistoryCache.size() > CACHE_SIZE) {
         Iterator<Map.Entry<String, JobInfo>> it = 
           jobHistoryCache.entrySet().iterator();
@@ -458,7 +478,102 @@ class JSPUtil {
         it.remove();
         LOG.info("Job History file removed form cache "+removeJobId);
       }
-      return jobInfo;
+    }
+
+    jobTracker.getJobACLsManager().checkAccess(JobID.forName(jobid),
+        UserGroupInformation.getCurrentUser(), JobACL.VIEW_JOB,
+        jobInfo.get(Keys.USER), jobInfo.getJobACLs().get(JobACL.VIEW_JOB));
+    return jobInfo;
+  }
+
+  /**
+   * Check the access for users to view job-history pages.
+   * 
+   * @param request
+   * @param response
+   * @param jobTracker
+   * @param fs
+   * @param logFile
+   * @return the job if authorization is disabled or if the authorization checks
+   *         pass. Otherwise return null.
+   * @throws IOException
+   * @throws InterruptedException
+   * @throws ServletException
+   */
+  static JobInfo checkAccessAndGetJobInfo(HttpServletRequest request,
+      HttpServletResponse response, final JobTracker jobTracker,
+      final FileSystem fs, final Path logFile) throws IOException,
+      InterruptedException, ServletException {
+    String jobid = getJobID(logFile.getName());
+    String user = request.getRemoteUser();
+    JobInfo job = null;
+    if (user != null) {
+      try {
+        final UserGroupInformation ugi =
+            UserGroupInformation.createRemoteUser(user);
+        job =
+            ugi.doAs(new PrivilegedExceptionAction<JobHistory.JobInfo>() {
+              public JobInfo run() throws IOException {
+                // checks job view permission
+                JobInfo jobInfo = JSPUtil.getJobInfo(logFile, fs, jobTracker);
+                return jobInfo;
+              }
+            });
+      } catch (AccessControlException e) {
+        String errMsg =
+            String.format(
+                "User %s failed to view %s!<br><br>%s"
+                    + "<hr>"
+                    + "<a href=\"jobhistory.jsp\">Go back to JobHistory</a><br>"
+                    + "<a href=\"jobtracker.jsp\">Go back to JobTracker</a>",
+                user, jobid, e.getMessage());
+        JSPUtil.setErrorAndForward(errMsg, request, response);
+        return null;
+      }
+    } else {
+      // no authorization needed
+      job = JSPUtil.getJobInfo(logFile, fs, jobTracker);
+    }
+    return job;
+  }
+
+  static String getJobID(String historyFileName) {
+    String[] jobDetails = historyFileName.split("_");
+    return jobDetails[2] + "_" + jobDetails[3] + "_" + jobDetails[4];
+  }
+
+  static String getUserName(String historyFileName) {
+    String[] jobDetails = historyFileName.split("_");
+    return jobDetails[5];
+  }
+
+  static String getJobName(String historyFileName) {
+    String[] jobDetails = historyFileName.split("_");
+    return jobDetails[6];
+  }
+
+  /**
+   * Nicely print the Job-ACLs
+   * @param tracker
+   * @param jobAcls
+   * @param out
+   * @throws IOException
+   */
+  static void printJobACLs(JobTracker tracker,
+      Map<JobACL, AccessControlList> jobAcls, JspWriter out)
+      throws IOException {
+    if (tracker.isJobLevelAuthorizationEnabled()) {
+      // Display job-view-acls and job-modify-acls configured for this job
+      out.print("<b>Job-ACLs:</b><br>");
+      for (JobACL aclName : JobACL.values()) {
+        String aclConfigName = aclName.getAclName();
+        AccessControlList aclConfigured = jobAcls.get(aclName);
+        if (aclConfigured != null) {
+          String aclStr = aclConfigured.toString();
+          out.print("&nbsp;&nbsp;&nbsp;&nbsp;" + aclConfigName + ": "
+              + aclStr + "<br>");
+        }
+      }
     }
   }
 

src/mapred/org/apache/hadoop/mapred/JobHistory.java

@@ -54,7 +54,9 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.mapreduce.JobACL;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authorize.AccessControlList;
 import org.apache.hadoop.util.StringUtils;
 
 /**
@@ -298,7 +300,8 @@ public class JobHistory {
     FINISHED_MAPS, FINISHED_REDUCES, JOB_STATUS, TASKID, HOSTNAME, TASK_TYPE, 
     ERROR, TASK_ATTEMPT_ID, TASK_STATUS, COPY_PHASE, SORT_PHASE, REDUCE_PHASE, 
     SHUFFLE_FINISHED, SORT_FINISHED, COUNTERS, SPLITS, JOB_PRIORITY, HTTP_PORT, 
-    TRACKER_NAME, STATE_STRING, VERSION, MAP_COUNTERS, REDUCE_COUNTERS
+    TRACKER_NAME, STATE_STRING, VERSION, MAP_COUNTERS, REDUCE_COUNTERS,
+    VIEW_JOB, MODIFY_JOB
   }
 
   /**
@@ -680,6 +683,8 @@ public class JobHistory {
   public static class JobInfo extends KeyValuePair{
     
     private Map<String, Task> allTasks = new TreeMap<String, Task>();
+    private Map<JobACL, AccessControlList> jobACLs =
+        new HashMap<JobACL, AccessControlList>();
     
     /** Create new JobInfo */
     public JobInfo(String jobId){ 
@@ -690,7 +695,26 @@ public class JobHistory {
      * Returns all map and reduce tasks <taskid-Task>. 
      */
     public Map<String, Task> getAllTasks() { return allTasks; }
-    
+
+    public Map<JobACL, AccessControlList> getJobACLs() {
+      return jobACLs;
+    }
+
+    @Override
+    public synchronized void handle(Map<Keys, String> values) {
+      // construct the ACLs
+      String viewJobACL = values.get(Keys.VIEW_JOB);
+      String modifyJobACL = values.get(Keys.MODIFY_JOB);
+      if (viewJobACL != null) {
+        jobACLs.put(JobACL.VIEW_JOB, new AccessControlList(viewJobACL));
+      }
+      if (modifyJobACL != null) {
+        jobACLs.put(JobACL.MODIFY_JOB, new AccessControlList(modifyJobACL));
+
+      }
+      super.handle(values);
+    }
+
     /**
      * Get the path of the locally stored job file
      * @param jobId id of the job
@@ -1210,9 +1234,12 @@ public class JobHistory {
 
           //add to writer as well 
           JobHistory.log(writers, RecordTypes.Job, 
-                         new Keys[]{Keys.JOBID, Keys.JOBNAME, Keys.USER, Keys.SUBMIT_TIME, Keys.JOBCONF }, 
+                         new Keys[]{Keys.JOBID, Keys.JOBNAME, Keys.USER, Keys.SUBMIT_TIME, Keys.JOBCONF, 
+                                      Keys.VIEW_JOB, Keys.MODIFY_JOB }, 
                          new String[]{jobId.toString(), jobName, user, 
-                                      String.valueOf(submitTime) , jobConfPath}
+                                      String.valueOf(submitTime) , jobConfPath,
+                                      jobConf.get(JobACL.VIEW_JOB.getAclName(), ""),
+                                      jobConf.get(JobACL.MODIFY_JOB.getAclName(), "")}
                         ); 
              
         }catch(IOException e){

src/mapred/org/apache/hadoop/mapred/TaskLogServlet.java

@@ -232,8 +232,10 @@ public class TaskLogServlet extends HttpServlet {
                      TaskLog.LogName.STDOUT, isCleanup);
         printTaskLog(response, out, attemptId, start, end, plainText, 
                      TaskLog.LogName.STDERR, isCleanup);
-        printTaskLog(response, out, attemptId, start, end, plainText, 
-                     TaskLog.LogName.SYSLOG, isCleanup);
+        if (haveTaskLog(attemptId, TaskLog.LogName.SYSLOG)) {
+          printTaskLog(response, out, attemptId, start, end, plainText,
+              TaskLog.LogName.SYSLOG, isCleanup);
+        }
         if (haveTaskLog(attemptId, TaskLog.LogName.DEBUGOUT)) {
           printTaskLog(response, out, attemptId, start, end, plainText, 
                        TaskLog.LogName.DEBUGOUT, isCleanup);

src/test/org/apache/hadoop/mapred/TestJobHistory.java

@@ -38,6 +38,7 @@ import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.mapred.JobHistory.*;
+import org.apache.hadoop.mapreduce.JobACL;
 import org.apache.hadoop.mapreduce.TaskType;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -798,6 +799,14 @@ public class TestJobHistory extends TestCase {
     validateJobLevelKeyValues(mr, job, jobInfo, conf);
     validateTaskLevelKeyValues(mr, job, jobInfo);
     validateTaskAttemptLevelKeyValues(mr, job, jobInfo);
+
+    // Also JobACLs should be correct
+    if (mr.getJobTrackerRunner().getJobTracker().isJobLevelAuthorizationEnabled()) {
+      assertEquals(conf.get(JobACL.VIEW_JOB.getAclName()),
+          jobInfo.getJobACLs().get(JobACL.VIEW_JOB).toString());
+      assertEquals(conf.get(JobACL.MODIFY_JOB.getAclName()),
+          jobInfo.getJobACLs().get(JobACL.MODIFY_JOB).toString());
+    }
   }
 
   public void testDoneFolderOnHDFS() throws IOException {
@@ -900,6 +909,9 @@ public class TestJobHistory extends TestCase {
       //set the done folder location
       String doneFolder = TEST_ROOT_DIR + "history_done";
       conf.set("mapred.job.tracker.history.completed.location", doneFolder);
+
+      // Enable ACLs so that they are logged to history
+      conf.setBoolean(JobConf.JOB_LEVEL_AUTHORIZATION_ENABLING_FLAG, true);
       
       mr = new MiniMRCluster(2, "file:///", 3, null, null, conf);
 
@@ -915,6 +927,8 @@ public class TestJobHistory extends TestCase {
 
       //Disable speculative execution
       conf.setSpeculativeExecution(false);
+      conf.set(JobACL.VIEW_JOB.getAclName(), "user1,user2 group1,group2");
+      conf.set(JobACL.MODIFY_JOB.getAclName(), "user3,user4 group3,group4");
 
       // Make sure that the job is not removed from memory until we do finish
       // the validation of history file content
@@ -983,7 +997,7 @@ public class TestJobHistory extends TestCase {
 
   //Returns the file in the done folder
   //Waits for sometime to get the file moved to done
-  private static String getDoneFile(JobConf conf, JobID id, 
+  static String getDoneFile(JobConf conf, JobID id, 
       Path doneDir) throws IOException {
     String name = null;
     for (int i = 0; name == null && i < 20; i++) {

src/test/org/apache/hadoop/mapred/TestWebUIAuthorization.java

@@ -17,17 +17,22 @@
  */
 package org.apache.hadoop.mapred;
 
+import java.io.File;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.net.URL;
 import java.net.HttpURLConnection;
+import java.net.URLEncoder;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.http.TestHttpServer.DummyFilterInitializer;
+import org.apache.hadoop.mapred.JobHistory.Keys;
+import org.apache.hadoop.mapred.JobHistory.TaskAttempt;
 import org.apache.hadoop.mapreduce.JobContext;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.examples.SleepJob;
@@ -42,6 +47,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Map.Entry;
 
 public class TestWebUIAuthorization extends ClusterMapReduceTestCase {
 
@@ -109,21 +115,28 @@ public class TestWebUIAuthorization extends ClusterMapReduceTestCase {
    *     cannot view the job
    * (5) other unauthorized users cannot view the job
    */
-  private void validateViewJob(String url, String method) throws IOException {
-    assertEquals(HttpURLConnection.HTTP_OK,
-        getHttpStatusCode(url, jobSubmitter, method));
-    assertEquals(HttpURLConnection.HTTP_OK,
-        getHttpStatusCode(url, superGroupMember, method));
-    assertEquals(HttpURLConnection.HTTP_OK,
-        getHttpStatusCode(url, mrOwner, method));
-    assertEquals(HttpURLConnection.HTTP_OK,
-        getHttpStatusCode(url, viewColleague, method));
-    assertEquals(HttpURLConnection.HTTP_OK,
-        getHttpStatusCode(url, viewAndModifyColleague, method));
-    assertEquals(HttpURLConnection.HTTP_UNAUTHORIZED,
-        getHttpStatusCode(url, modifyColleague, method));
-    assertEquals(HttpURLConnection.HTTP_UNAUTHORIZED,
-        getHttpStatusCode(url, unauthorizedUser, method));
+  private void validateViewJob(String url, String method)
+      throws IOException {
+    assertEquals("Incorrect return code for " + jobSubmitter,
+        HttpURLConnection.HTTP_OK, getHttpStatusCode(url, jobSubmitter,
+            method));
+    assertEquals("Incorrect return code for " + superGroupMember,
+        HttpURLConnection.HTTP_OK, getHttpStatusCode(url, superGroupMember,
+            method));
+    assertEquals("Incorrect return code for " + mrOwner,
+        HttpURLConnection.HTTP_OK, getHttpStatusCode(url, mrOwner, method));
+    assertEquals("Incorrect return code for " + viewColleague,
+        HttpURLConnection.HTTP_OK, getHttpStatusCode(url, viewColleague,
+            method));
+    assertEquals("Incorrect return code for " + viewAndModifyColleague,
+        HttpURLConnection.HTTP_OK, getHttpStatusCode(url,
+            viewAndModifyColleague, method));
+    assertEquals("Incorrect return code for " + modifyColleague,
+        HttpURLConnection.HTTP_UNAUTHORIZED, getHttpStatusCode(url,
+            modifyColleague, method));
+    assertEquals("Incorrect return code for " + unauthorizedUser,
+        HttpURLConnection.HTTP_UNAUTHORIZED, getHttpStatusCode(url,
+            unauthorizedUser, method));
   }
 
   /**
@@ -221,6 +234,137 @@ public class TestWebUIAuthorization extends ClusterMapReduceTestCase {
     }
   }
 
+  public void testAuthorizationForJobHistoryPages() throws Exception {
+    JobConf conf = new JobConf();
+    conf.set(CommonConfigurationKeys.HADOOP_SECURITY_GROUP_MAPPING,
+        MyGroupsProvider.class.getName());
+    Groups.getUserToGroupsMappingService(conf);
+    Properties props = new Properties();
+    props.setProperty("hadoop.http.filter.initializers",
+        DummyFilterInitializer.class.getName());
+    props.setProperty(JobConf.JOB_LEVEL_AUTHORIZATION_ENABLING_FLAG,
+        String.valueOf(true));
+    props.setProperty("dfs.permissions", "false");
+
+    props.setProperty("mapreduce.job.committer.setup.cleanup.needed",
+        "false");
+    props.setProperty(JobConf.MR_SUPERGROUP, "superGroup");
+
+    MyGroupsProvider.mapping.put(jobSubmitter, Arrays.asList("group1"));
+    MyGroupsProvider.mapping.put(viewColleague, Arrays.asList("group2"));
+    MyGroupsProvider.mapping.put(modifyColleague, Arrays.asList("group1"));
+    MyGroupsProvider.mapping.put(unauthorizedUser, Arrays.asList("evilSociety"));
+    MyGroupsProvider.mapping.put(superGroupMember, Arrays.asList("superGroup"));
+    MyGroupsProvider.mapping.put(viewAndModifyColleague, Arrays.asList("group3"));
+    mrOwner = UserGroupInformation.getCurrentUser().getShortUserName();
+    MyGroupsProvider.mapping.put(mrOwner, Arrays.asList(
+        new String[] { "group4", "group5" }));
+
+    startCluster(true, props);
+    MiniMRCluster cluster = getMRCluster();
+    int infoPort = cluster.getJobTrackerRunner().getJobTrackerInfoPort();
+
+    conf = new JobConf(cluster.createJobConf());
+    conf.set(JobContext.JOB_ACL_VIEW_JOB, viewColleague + " group3");
+
+    // Let us add group1 and group3 to modify-job-acl. So modifyColleague and
+    // viewAndModifyColleague will be able to modify the job
+    conf.set(JobContext.JOB_ACL_MODIFY_JOB, " group1,group3");
+
+    final SleepJob sleepJob = new SleepJob();
+    final JobConf jobConf = new JobConf(conf);
+    sleepJob.setConf(jobConf);
+    UserGroupInformation jobSubmitterUGI =
+        UserGroupInformation.createRemoteUser(jobSubmitter);
+    RunningJob job =
+        jobSubmitterUGI.doAs(new PrivilegedExceptionAction<RunningJob>() {
+          public RunningJob run() throws Exception {
+            JobClient jobClient = new JobClient(jobConf);
+            SleepJob sleepJob = new SleepJob();
+            sleepJob.setConf(jobConf);
+            JobConf jobConf =
+                sleepJob.setupJobConf(1, 0, 1000, 1, 1000, 1000);
+            RunningJob runningJob = jobClient.runJob(jobConf);
+            return runningJob;
+          }
+        });
+
+    JobID jobid = job.getID();
+
+    JobTracker jobTracker = getMRCluster().getJobTrackerRunner().getJobTracker();
+    JobInProgress jip = jobTracker.getJob(jobid);
+    JobConf finalJobConf = jip.getJobConf();
+    Path doneDir = JobHistory.getCompletedJobHistoryLocation();
+
+    // Wait for history file to be written
+    while(TestJobHistory.getDoneFile(finalJobConf, jobid, doneDir) == null) {
+      Thread.sleep(2000);
+    }
+
+    Path historyFilePath = new Path(doneDir,
+        JobHistory.JobInfo.getDoneJobHistoryFileName(finalJobConf, jobid));
+
+    String urlEncodedHistoryFileName = URLEncoder.encode(historyFilePath.toString());
+    String jtURL = "http://localhost:" + infoPort;
+
+    // validate access of jobdetails_history.jsp
+    String jobDetailsJSP =
+        jtURL + "/jobdetailshistory.jsp?logFile=" + urlEncodedHistoryFileName;
+    validateViewJob(jobDetailsJSP, "GET");
+
+    // validate accesses of jobtaskshistory.jsp
+    String jobTasksJSP =
+        jtURL + "/jobtaskshistory.jsp?logFile=" + urlEncodedHistoryFileName;
+    String[] taskTypes =
+        new String[] { "JOb_SETUP", "MAP", "REDUCE", "JOB_CLEANUP" };
+    String[] states =
+        new String[] { "all", "SUCCEEDED", "FAILED", "KILLED" };
+    for (String taskType : taskTypes) {
+      for (String state : states) {
+        validateViewJob(jobTasksJSP + "&taskType=" + taskType + "&status="
+            + state, "GET");
+      }
+    }
+
+    JobHistory.JobInfo jobInfo = new JobHistory.JobInfo(jobid.toString());
+
+    DefaultJobHistoryParser.JobTasksParseListener l =
+                   new DefaultJobHistoryParser.JobTasksParseListener(jobInfo);
+    JobHistory.parseHistoryFromFS(historyFilePath.toString().substring(5),
+        l, historyFilePath.getFileSystem(conf));
+
+    Map<String, org.apache.hadoop.mapred.JobHistory.Task> tipsMap = jobInfo.getAllTasks();
+    for (String tip : tipsMap.keySet()) {
+      // validate access of taskdetailshistory.jsp
+      validateViewJob(jtURL + "/taskdetailshistory.jsp?logFile="
+          + urlEncodedHistoryFileName + "&tipid=" + tip.toString(), "GET");
+
+      Map<String, TaskAttempt> attemptsMap =
+          tipsMap.get(tip).getTaskAttempts();
+      for (String attempt : attemptsMap.keySet()) {
+
+        // validate access to taskstatshistory.jsp
+        validateViewJob(jtURL + "/taskstatshistory.jsp?attemptid="
+            + attempt.toString() + "&logFile=" + urlEncodedHistoryFileName, "GET");
+
+        // validate access to tasklogs
+        validateViewJob(TaskLogServlet.getTaskLogUrl("localhost",
+            attemptsMap.get(attempt).get(Keys.HTTP_PORT),
+            attempt.toString()), "GET");
+      }
+    }
+
+    // validate access to analysejobhistory.jsp
+    String analyseJobHistoryJSP =
+        jtURL + "/analysejobhistory.jsp?logFile=" + urlEncodedHistoryFileName;
+    validateViewJob(analyseJobHistoryJSP, "GET");
+
+    // validate access of jobconf_history.jsp
+    String jobConfJSP =
+        jtURL + "/jobconf_history.jsp?logFile=" + urlEncodedHistoryFileName;
+    validateViewJob(jobConfJSP, "GET");
+  }
+
   /**
    * Starts a sleep job and tries to kill the job using jobdetails.jsp as
    * (1) viewColleague (2) unauthorizedUser (3) modifyColleague

src/webapps/job/analysejobhistory.jsp

@@ -16,18 +16,26 @@
 %>
 <html><body>
 <%
-  String jobid = JobID.forName(request.getParameter("jobid")).toString();
   String logFile = request.getParameter("logFile");
+  if (logFile == null) {
+    out.println("Missing job!!");
+    return;
+  }
   String encodedLogFileName = JobHistory.JobInfo.encodeJobHistoryFilePath(logFile);
+  String jobid = JSPUtil.getJobID(new Path(encodedLogFileName).getName());
   String numTasks = request.getParameter("numTasks");
   int showTasks = 10 ; 
   if (numTasks != null) {
     showTasks = Integer.parseInt(numTasks);  
   }
   FileSystem fs = (FileSystem) application.getAttribute("fileSys");
-  JobHistory.JobInfo job = JSPUtil.getJobInfo(request, fs);
-%>
-<h2>Hadoop Job <a href="jobdetailshistory.jsp?jobid=<%=jobid%>&&logFile=<%=encodedLogFileName%>"><%=jobid %> </a></h2>
+  JobTracker jobTracker = (JobTracker) application.getAttribute("job.tracker");
+  JobHistory.JobInfo job = JSPUtil.checkAccessAndGetJobInfo(request,
+      response, jobTracker, fs, new Path(logFile));
+  if (job == null) {
+    return;
+  }%>
+<h2>Hadoop Job <a href="jobdetailshistory.jsp?logFile=<%=encodedLogFileName%>"><%=jobid %> </a></h2>
 <b>User : </b> <%=HtmlQuoting.quoteHtmlChars(job.get(Keys.USER)) %><br/> 
 <b>JobName : </b> <%=HtmlQuoting.quoteHtmlChars(job.get(Keys.JOBNAME)) %><br/> 
 <b>JobConf : </b> <%=job.get(Keys.JOBCONF) %><br/> 
@@ -101,7 +109,7 @@
 %>
 
 <h3>Time taken by best performing Map task 
-<a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>&taskid=<%=minMap.get(Keys.TASKID)%>">
+<a href="taskdetailshistory.jsp?logFile=<%=encodedLogFileName%>&tipid=<%=minMap.get(Keys.TASKID)%>">
 <%=minMap.get(Keys.TASKID) %></a> : <%=StringUtils.formatTimeDiff(minMap.getLong(Keys.FINISH_TIME), minMap.getLong(Keys.START_TIME) ) %></h3>
 <h3>Average time taken by Map tasks: 
 <%=StringUtils.formatTimeDiff(avgMapTime, 0) %></h3>
@@ -112,7 +120,7 @@
   for (int i=0;i<showTasks && i<mapTasks.length; i++) {
 %>
     <tr>
-    <td><a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>&taskid=<%=mapTasks[i].get(Keys.TASKID)%>">
+    <td><a href="taskdetailshistory.jsp?logFile=<%=encodedLogFileName%>&tipid=<%=mapTasks[i].get(Keys.TASKID)%>">
         <%=mapTasks[i].get(Keys.TASKID) %></a></td>
     <td><%=StringUtils.formatTimeDiff(mapTasks[i].getLong(Keys.FINISH_TIME), mapTasks[i].getLong(Keys.START_TIME)) %></td>
     </tr>
@@ -134,8 +142,8 @@
 %>
 
 <h3>The last Map task 
-<a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>
-&taskid=<%=lastMap.get(Keys.TASKID)%>"><%=lastMap.get(Keys.TASKID) %></a> 
+<a href="taskdetailshistory.jsp?logFile=<%=encodedLogFileName%>
+&tipid=<%=lastMap.get(Keys.TASKID)%>"><%=lastMap.get(Keys.TASKID) %></a> 
 finished at (relative to the Job launch time): 
 <%=StringUtils.getFormattedTimeWithDiff(dateFormat, 
                               lastMap.getLong(Keys.FINISH_TIME), 
@@ -147,9 +155,9 @@ finished at (relative to the Job launch time):
   Arrays.sort(reduceTasks, cShuffle); 
   JobHistory.Task minShuffle = reduceTasks[reduceTasks.length-1] ;
 %>
-<h3>Time taken by best performing shuffle
-<a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>
-&taskid=<%=minShuffle.get(Keys.TASKID)%>"><%=minShuffle.get(Keys.TASKID)%></a> : 
+<h3>Time taken by best performing shufflejobId
+<a href="taskdetailshistory.jsp?logFile=<%=encodedLogFileName%>
+&tipid=<%=minShuffle.get(Keys.TASKID)%>"><%=minShuffle.get(Keys.TASKID)%></a> : 
 <%=StringUtils.formatTimeDiff(minShuffle.getLong(Keys.SHUFFLE_FINISHED), 
                               minShuffle.getLong(Keys.START_TIME) ) %></h3>
 <h3>Average time taken by Shuffle: 
@@ -161,8 +169,8 @@ finished at (relative to the Job launch time):
   for (int i=0;i<showTasks && i<reduceTasks.length; i++) {
 %>
     <tr>
-    <td><a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=
-<%=encodedLogFileName%>&taskid=<%=reduceTasks[i].get(Keys.TASKID)%>">
+    <td><a href="taskdetailshistory.jsp?logFile=
+<%=encodedLogFileName%>&tipid=<%=reduceTasks[i].get(Keys.TASKID)%>">
 <%=reduceTasks[i].get(Keys.TASKID) %></a></td>
     <td><%=
            StringUtils.formatTimeDiff(
@@ -188,8 +196,8 @@ finished at (relative to the Job launch time):
 %>
 
 <h3>The last Shuffle  
-<a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>
-&taskid=<%=lastShuffle.get(Keys.TASKID)%>"><%=lastShuffle.get(Keys.TASKID)%>
+<a href="taskdetailshistory.jsp?logFile=<%=encodedLogFileName%>
+&tipid=<%=lastShuffle.get(Keys.TASKID)%>"><%=lastShuffle.get(Keys.TASKID)%>
 </a> finished at (relative to the Job launch time): 
 <%=StringUtils.getFormattedTimeWithDiff(dateFormat,
                               lastShuffle.getLong(Keys.SHUFFLE_FINISHED), 
@@ -210,7 +218,7 @@ finished at (relative to the Job launch time):
 %>
 <hr/>
 <h3>Time taken by best performing Reduce task : 
-<a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>&taskid=<%=minReduce.get(Keys.TASKID)%>">
+<a href="taskdetailshistory.jsp?logFile=<%=encodedLogFileName%>&tipid=<%=minReduce.get(Keys.TASKID)%>">
 <%=minReduce.get(Keys.TASKID) %></a> : 
 <%=StringUtils.formatTimeDiff(minReduce.getLong(Keys.FINISH_TIME),
     minReduce.getLong(Keys.SHUFFLE_FINISHED) ) %></h3>
@@ -224,7 +232,7 @@ finished at (relative to the Job launch time):
   for (int i=0;i<showTasks && i<reduceTasks.length; i++) {
 %>
     <tr>
-    <td><a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>&taskid=<%=reduceTasks[i].get(Keys.TASKID)%>">
+    <td><a href="taskdetailshistory.jsp?logFile=<%=encodedLogFileName%>&tipid=<%=reduceTasks[i].get(Keys.TASKID)%>">
         <%=reduceTasks[i].get(Keys.TASKID) %></a></td>
     <td><%=StringUtils.formatTimeDiff(
              reduceTasks[i].getLong(Keys.FINISH_TIME), 
@@ -240,8 +248,8 @@ finished at (relative to the Job launch time):
 %>
 
 <h3>The last Reduce task 
-<a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>
-&taskid=<%=lastReduce.get(Keys.TASKID)%>"><%=lastReduce.get(Keys.TASKID)%>
+<a href="taskdetailshistory.jsp?logFile=<%=encodedLogFileName%>
+&tipid=<%=lastReduce.get(Keys.TASKID)%>"><%=lastReduce.get(Keys.TASKID)%>
 </a> finished at (relative to the Job launch time): 
 <%=StringUtils.getFormattedTimeWithDiff(dateFormat,
                               lastReduce.getLong(Keys.FINISH_TIME), 

src/webapps/job/jobconf_history.jsp

@@ -6,16 +6,25 @@
   import="org.apache.hadoop.mapred.*"
   import="org.apache.hadoop.fs.*"
   import="org.apache.hadoop.util.*"
+  import="org.apache.hadoop.mapreduce.JobACL"
+  import="org.apache.hadoop.security.UserGroupInformation"
+  import="org.apache.hadoop.security.authorize.AccessControlList"
+  import="org.apache.hadoop.security.AccessControlException"
 %>
 
 
 <%
   JobTracker tracker = (JobTracker) application.getAttribute("job.tracker");
-  String jobId = JobID.forName(request.getParameter("jobid")).toString();
-  if (jobId == null) {
-    out.println("<h2>Missing 'jobid' for fetching job configuration!</h2>");
- 	return;
+
+  String logFileString = request.getParameter("logFile");
+  if (logFileString == null) {
+    out.println("<h2>Missing 'logFile' for fetching job configuration!</h2>");
+    return;
   }
+
+  Path logFile = new Path(logFileString);
+  String jobId = JSPUtil.getJobID(logFile.getName()).toString();
+
 %>
   
 <html>
@@ -26,14 +35,31 @@
 <h2>Job Configuration: JobId - <%= jobId %></h2><br>
 
 <%
-  Path logDir = new Path(request.getParameter("jobLogDir"));
-  Path jobFilePath = new Path(logDir, 
-                       request.getParameter("jobUniqueString") + "_conf.xml");
+  Path jobFilePath = JSPUtil.getJobConfFilePath(logFile);
   FileSystem fs = (FileSystem) application.getAttribute("fileSys");
   FSDataInputStream jobFile = null; 
   try {
     jobFile = fs.open(jobFilePath);
     JobConf jobConf = new JobConf(jobFilePath);
+    JobTracker jobTracker = (JobTracker) application.getAttribute("job.tracker");
+    String user = request.getRemoteUser();
+    if (user != null) {
+      try {
+        jobTracker.getJobACLsManager().checkAccess(JobID.forName(jobId),
+            UserGroupInformation.createRemoteUser(user), JobACL.VIEW_JOB,
+            jobConf.getUser(),
+            new AccessControlList(jobConf.get(JobACL.VIEW_JOB.getAclName())));
+        } catch (AccessControlException e) {
+          String errMsg =
+            user
+                + " is not authorized to view details of job "
+                + jobId
+                + "<hr><a href=\"jobhistory.jsp\">Go back to JobHistory</a><br>";
+        JSPUtil.setErrorAndForward(errMsg, request, response);
+        return;
+        }
+    }
+
     XMLUtils.transform(
         jobConf.getConfResourceAsInputStream("webapps/static/jobconf.xsl"),
         jobFile, out);

src/webapps/job/jobdetails.jsp

@@ -265,8 +265,8 @@
         out.println("<h2>Job " + jobId + " not known!</h2>");
         return;
       }
-      String historyUrl = "/jobdetailshistory.jsp?jobid=" + jobId +
-      "&logFile=" + JobHistory.JobInfo.encodeJobHistoryFilePath(historyFile);
+      String historyUrl = "/jobdetailshistory.jsp?logFile=" + 
+          JobHistory.JobInfo.encodeJobHistoryFilePath(historyFile);
       response.sendRedirect(response.encodeRedirectURL(historyUrl));
       return;
     }
@@ -281,21 +281,8 @@
     out.print("<b>Job File:</b> <a href=\"jobconf.jsp?jobid=" + jobId + "\">" +
         profile.getJobFile() + "</a><br>\n");
 
-    if (tracker.isJobLevelAuthorizationEnabled()) {
-      // Display job-view-acls and job-modify-acls configured for this job
-      Map<JobACL, AccessControlList> jobAcls = status.getJobACLs();
-      out.print("<b>Job-ACLs:</b><br>");
-      for (JobACL aclName : JobACL.values()) {
-        String aclConfigName = aclName.getAclName();
-        AccessControlList aclConfigured = jobAcls.get(aclName);
-        String aclStr = "";
-        if (aclConfigured != null) {
-          aclStr = aclConfigured.toString();
-        }
-        out.print("&nbsp;&nbsp;&nbsp;&nbsp;" + aclConfigName + ": "
-                  + aclStr + "<br>");
-      }
-    }
+    Map<JobACL, AccessControlList> jobAcls = status.getJobACLs();
+    JSPUtil.printJobACLs(tracker, jobAcls, out);
     out.print("<b>Job Setup:</b>");
     printJobLevelTaskSummary(out, jobId, "setup", 
                              job.getTasks(TaskType.JOB_SETUP));

src/webapps/job/jobdetailshistory.jsp

@@ -9,20 +9,27 @@
   import="org.apache.hadoop.util.*"
   import="java.text.*"
   import="org.apache.hadoop.mapred.JobHistory.*"
+  import="java.security.PrivilegedExceptionAction"
+  import="org.apache.hadoop.security.AccessControlException"
+  import="org.apache.hadoop.mapreduce.JobACL"
+  import="org.apache.hadoop.security.authorize.AccessControlList"
 %>
 
 <%! static SimpleDateFormat dateFormat = new SimpleDateFormat("d-MMM-yyyy HH:mm:ss") ; %>
 <%
-    String jobid = JobID.forName(request.getParameter("jobid")).toString();
     String logFile = request.getParameter("logFile");
-	String encodedLogFileName = JobHistory.JobInfo.encodeJobHistoryFilePath(logFile);
-	
-    Path jobFile = new Path(logFile);
-    String[] jobDetails = jobFile.getName().split("_");
-    String jobUniqueString = jobDetails[0] + "_" +jobDetails[1] + "_" + jobid ;
+
+    String jobid = JSPUtil.getJobID(new Path(logFile).getName());
 	
     FileSystem fs = (FileSystem) application.getAttribute("fileSys");
-    JobHistory.JobInfo job = JSPUtil.getJobInfo(request, fs);
+    JobTracker jobTracker = (JobTracker) application.getAttribute("job.tracker");
+    JobHistory.JobInfo job = JSPUtil.checkAccessAndGetJobInfo(request,
+        response, jobTracker, fs, new Path(logFile));
+    if (job == null) {
+      return;
+    }
+
+  	String encodedLogFileName = JobHistory.JobInfo.encodeJobHistoryFilePath(logFile);
 %>
 
 <html>
@@ -35,9 +42,13 @@
 <h2>Hadoop Job <%=jobid %> on <a href="jobhistory.jsp">History Viewer</a></h2>
 
 <b>User: </b> <%=HtmlQuoting.quoteHtmlChars(job.get(Keys.USER)) %><br/> 
-<b>JobName: </b> <%=HtmlQuoting.quoteHtmlChars(job.get(Keys.JOBNAME)) %><br/> 
-<b>JobConf: </b> <a href="jobconf_history.jsp?jobid=<%=jobid%>&jobLogDir=<%=new Path(logFile).getParent().toString()%>&jobUniqueString=<%=jobUniqueString%>"> 
+<b>JobName: </b> <%=HtmlQuoting.quoteHtmlChars(job.get(Keys.JOBNAME)) %><br/>  
+<b>JobConf: </b> <a href="jobconf_history.jsp?logFile=<%=encodedLogFileName%>"> 
                  <%=job.get(Keys.JOBCONF) %></a><br/> 
+<%         
+  Map<JobACL, AccessControlList> jobAcls = job.getJobACLs();
+  JSPUtil.printJobACLs(jobTracker, jobAcls, out);
+%> 
 <b>Submitted At: </b> <%=StringUtils.getFormattedTimeWithDiff(dateFormat, job.getLong(Keys.SUBMIT_TIME), 0 )  %><br/> 
 <b>Launched At: </b> <%=StringUtils.getFormattedTimeWithDiff(dateFormat, job.getLong(Keys.LAUNCH_TIME), job.getLong(Keys.SUBMIT_TIME)) %><br/>
 <b>Finished At: </b>  <%=StringUtils.getFormattedTimeWithDiff(dateFormat, job.getLong(Keys.FINISH_TIME), job.getLong(Keys.LAUNCH_TIME)) %><br/>
@@ -135,7 +146,7 @@
       }
     }
 %>
-<b><a href="analysejobhistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>">Analyse This Job</a></b> 
+<b><a href="analysejobhistory.jsp?logFile=<%=encodedLogFileName%>">Analyse This Job</a></b> 
 <hr/>
 <center>
 <table border="2" cellpadding="5" cellspacing="2">
@@ -144,52 +155,52 @@
 </tr>
 <tr>
 <td>Setup</td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.SETUP.name() %>&status=all">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.SETUP.name() %>&status=all">
         <%=totalSetups%></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.SETUP.name() %>&status=<%=Values.SUCCESS %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.SETUP.name() %>&status=<%=Values.SUCCESS %>">
         <%=numFinishedSetups%></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.SETUP.name() %>&status=<%=Values.FAILED %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.SETUP.name() %>&status=<%=Values.FAILED %>">
         <%=numFailedSetups%></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.SETUP.name() %>&status=<%=Values.KILLED %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.SETUP.name() %>&status=<%=Values.KILLED %>">
         <%=numKilledSetups%></a></td>  
     <td><%=StringUtils.getFormattedTimeWithDiff(dateFormat, setupStarted, 0) %></td>
     <td><%=StringUtils.getFormattedTimeWithDiff(dateFormat, setupFinished, setupStarted) %></td>
 </tr>
 <tr>
 <td>Map</td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.MAP.name() %>&status=all">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.MAP.name() %>&status=all">
         <%=totalMaps %></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.MAP.name() %>&status=<%=Values.SUCCESS %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.MAP.name() %>&status=<%=Values.SUCCESS %>">
         <%=job.getInt(Keys.FINISHED_MAPS) %></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.MAP.name() %>&status=<%=Values.FAILED %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.MAP.name() %>&status=<%=Values.FAILED %>">
         <%=numFailedMaps %></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.MAP.name() %>&status=<%=Values.KILLED %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.MAP.name() %>&status=<%=Values.KILLED %>">
         <%=numKilledMaps %></a></td>
     <td><%=StringUtils.getFormattedTimeWithDiff(dateFormat, mapStarted, 0) %></td>
     <td><%=StringUtils.getFormattedTimeWithDiff(dateFormat, mapFinished, mapStarted) %></td>
 </tr>
 <tr>
 <td>Reduce</td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.REDUCE.name() %>&status=all">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.REDUCE.name() %>&status=all">
         <%=totalReduces%></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.REDUCE.name() %>&status=<%=Values.SUCCESS %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.REDUCE.name() %>&status=<%=Values.SUCCESS %>">
         <%=job.getInt(Keys.FINISHED_REDUCES)%></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.REDUCE.name() %>&status=<%=Values.FAILED %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.REDUCE.name() %>&status=<%=Values.FAILED %>">
         <%=numFailedReduces%></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.REDUCE.name() %>&status=<%=Values.KILLED %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.REDUCE.name() %>&status=<%=Values.KILLED %>">
         <%=numKilledReduces%></a></td>  
     <td><%=StringUtils.getFormattedTimeWithDiff(dateFormat, reduceStarted, 0) %></td>
     <td><%=StringUtils.getFormattedTimeWithDiff(dateFormat, reduceFinished, reduceStarted) %></td>
 </tr>
 <tr>
 <td>Cleanup</td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.CLEANUP.name() %>&status=all">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.CLEANUP.name() %>&status=all">
         <%=totalCleanups%></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.CLEANUP.name() %>&status=<%=Values.SUCCESS %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.CLEANUP.name() %>&status=<%=Values.SUCCESS %>">
         <%=numFinishedCleanups%></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.CLEANUP.name() %>&status=<%=Values.FAILED %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.CLEANUP.name() %>&status=<%=Values.FAILED %>">
         <%=numFailedCleanups%></a></td>
-    <td><a href="jobtaskshistory.jsp?jobid=<%=jobid %>&logFile=<%=encodedLogFileName%>&taskType=<%=Values.CLEANUP.name() %>&status=<%=Values.KILLED %>">
+    <td><a href="jobtaskshistory.jsp?logFile=<%=encodedLogFileName%>&taskType=<%=Values.CLEANUP.name() %>&status=<%=Values.KILLED %>">
         <%=numKilledCleanups%></a></td>  
     <td><%=StringUtils.getFormattedTimeWithDiff(dateFormat, cleanupStarted, 0) %></td>
     <td><%=StringUtils.getFormattedTimeWithDiff(dateFormat, cleanupFinished, cleanupStarted) %></td>
@@ -281,7 +292,7 @@
 <%
         for (String t : failedTasks) {
 %>
-          <a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>&taskid=<%=t %>"><%=t %></a>,&nbsp;
+          <a href="taskdetailshistory.jsp?logFile=<%=encodedLogFileName%>&tipid=<%=t %>"><%=t %></a>,&nbsp;
 <%		  
         }
 %>	
@@ -315,7 +326,7 @@
 <%
         for (String t : killedTasks) {
 %>
-          <a href="taskdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>&taskid=<%=t %>"><%=t %></a>,&nbsp;
+          <a href="taskdetailshistory.jsp?logFile=<%=encodedLogFileName%>&tipid=<%=t %>"><%=t %></a>,&nbsp;
 <%		  
         }
 %>	

src/webapps/job/jobhistory.jsp

@@ -264,8 +264,8 @@ window.location.href = url;
       out.print("<tr>"); 
       out.print("<td>" + trackerHostName + "</td>"); 
       out.print("<td>" + new Date(Long.parseLong(trackerid)) + "</td>"); 
-      out.print("<td>" + "<a href=\"jobdetailshistory.jsp?jobid=" + jobId + 
-                "&logFile=" + logFile.toString() + "\">" + jobId + "</a></td>"); 
+      out.print("<td>" + "<a href=\"jobdetailshistory.jsp?logFile=" 
+          + logFile.toString() + "\">" + jobId + "</a></td>");
       out.print("<td>" + HtmlQuoting.quoteHtmlChars(jobName) + "</td>"); 
       out.print("<td>" + HtmlQuoting.quoteHtmlChars(user) + "</td>"); 
       out.print("</tr>");

src/webapps/job/jobtaskshistory.jsp

@@ -17,19 +17,24 @@
 %>
 
 <%	
-  String jobid = JobID.forName(request.getParameter("jobid")).toString();
   String logFile = request.getParameter("logFile");
   String encodedLogFileName = JobHistory.JobInfo.encodeJobHistoryFilePath(logFile);
-  String taskStatus = request.getParameter("status");
+  String jobid = JSPUtil.getJobID(new Path(logFile).getName());
+  String taskStatus = request.getParameter("status"); 
   String taskType = request.getParameter("taskType"); 
   
   FileSystem fs = (FileSystem) application.getAttribute("fileSys");
-  JobHistory.JobInfo job = JSPUtil.getJobInfo(request, fs);
+  JobTracker jobTracker = (JobTracker) application.getAttribute("job.tracker");
+  JobHistory.JobInfo job = JSPUtil.checkAccessAndGetJobInfo(request,
+      response, jobTracker, fs, new Path(logFile));
+  if (job == null) {
+    return;
+  }
   Map<String, JobHistory.Task> tasks = job.getAllTasks(); 
 %>
 <html>
 <body>
-<h2><%=taskStatus%> <%=taskType %> task list for <a href="jobdetailshistory.jsp?jobid=<%=jobid%>&&logFile=<%=encodedLogFileName%>"><%=jobid %> </a></h2>
+<h2><%=taskStatus%> <%=taskType %> task list for <a href="jobdetailshistory.jsp?logFile=<%=encodedLogFileName%>"><%=jobid %> </a></h2>
 <center>
 <table border="2" cellpadding="5" cellspacing="2">
 <tr><td>Task Id</td><td>Start Time</td><td>Finish Time<br/></td><td>Error</td></tr>
@@ -40,7 +45,7 @@
       for (JobHistory.TaskAttempt taskAttempt : taskAttempts.values()) {
         if (taskStatus.equals(taskAttempt.get(Keys.TASK_STATUS)) || 
           taskStatus.equals("all")){
-          printTask(jobid, encodedLogFileName, taskAttempt, out); 
+          printTask(encodedLogFileName, taskAttempt, out); 
         }
       }
     }
@@ -48,11 +53,11 @@
 %>
 </table>
 <%!
-  private void printTask(String jobid, String logFile,
+  private void printTask(String logFile,
     JobHistory.TaskAttempt attempt, JspWriter out) throws IOException{
     out.print("<tr>"); 
-    out.print("<td>" + "<a href=\"taskdetailshistory.jsp?jobid=" + jobid + 
-          "&logFile="+ logFile +"&taskid="+attempt.get(Keys.TASKID)+"\">" +
+    out.print("<td>" + "<a href=\"taskdetailshistory.jsp?logFile="+ logFile
+     +"&tipid="+attempt.get(Keys.TASKID)+"\">" +
           attempt.get(Keys.TASKID) + "</a></td>");
     out.print("<td>" + StringUtils.getFormattedTimeWithDiff(dateFormat, 
           attempt.getLong(Keys.START_TIME), 0 ) + "</td>");

src/webapps/job/taskdetailshistory.jsp

@@ -14,18 +14,27 @@
 <%!	private static SimpleDateFormat dateFormat = new SimpleDateFormat("d/MM HH:mm:ss") ; %>
 
 <%	
-  String jobid = JobID.forName(request.getParameter("jobid")).toString();
   String logFile = request.getParameter("logFile");
+  String tipid = request.getParameter("tipid");
+  if (logFile == null || tipid == null) {
+    out.println("Missing job!!");
+    return;
+  }
   String encodedLogFileName = JobHistory.JobInfo.encodeJobHistoryFilePath(logFile);
-  String taskid = request.getParameter("taskid"); 
+  String jobid = JSPUtil.getJobID(new Path(encodedLogFileName).getName());
   FileSystem fs = (FileSystem) application.getAttribute("fileSys");
-  JobHistory.JobInfo job = JSPUtil.getJobInfo(request, fs);
-  JobHistory.Task task = job.getAllTasks().get(taskid); 
+  JobTracker jobTracker = (JobTracker) application.getAttribute("job.tracker");
+  JobHistory.JobInfo job = JSPUtil.checkAccessAndGetJobInfo(request,
+      response, jobTracker, fs, new Path(logFile));
+  if (job == null) {
+    return;
+  }
+  JobHistory.Task task = job.getAllTasks().get(tipid); 
   String type = task.get(Keys.TASK_TYPE);
 %>
 <html>
 <body>
-<h2><%=taskid %> attempts for <a href="jobdetailshistory.jsp?jobid=<%=jobid%>&&logFile=<%=encodedLogFileName%>"> <%=jobid %> </a></h2>
+<h2><%=tipid %> attempts for <a href="jobdetailshistory.jsp?logFile=<%=encodedLogFileName%>"> <%=jobid %> </a></h2>
 <center>
 <table border="2" cellpadding="5" cellspacing="2">
 <tr><td>Task Id</td><td>Start Time</td>
@@ -106,12 +115,10 @@
     if (counters != null) {
       TaskAttemptID attemptId = 
         TaskAttemptID.forName(taskAttempt.get(Keys.TASK_ATTEMPT_ID));
-      TaskID taskId = attemptId.getTaskID();
-      org.apache.hadoop.mapreduce.JobID jobId = taskId.getJobID();
+      TaskID tipid = attemptId.getTaskID();
+      org.apache.hadoop.mapreduce.JobID jobId = tipid.getJobID();
       out.print("<td>" 
-       + "<a href=\"/taskstatshistory.jsp?jobid=" + jobId
-           + "&taskid=" + taskId
-           + "&attemptid=" + attemptId
+       + "<a href=\"/taskstatshistory.jsp?attemptid=" + attemptId
            + "&logFile=" + logFile + "\">"
            + counters.size() + "</a></td>");
     } else {

src/webapps/job/taskstatshistory.jsp

@@ -34,19 +34,27 @@
 %>
 
 <%
-  String jobid = request.getParameter("jobid");
   String attemptid = request.getParameter("attemptid");
-  String taskid = request.getParameter("taskid");
+  if(attemptid == null) {
+    out.println("No attemptid found! Pass a 'attemptid' parameter in the request.");
+    return;
+  }
+  TaskID tipid = TaskAttemptID.forName(attemptid).getTaskID();
   String logFile = request.getParameter("logFile");
   String encodedLogFileName = 
     JobHistory.JobInfo.encodeJobHistoryFilePath(logFile);
-
+  String jobid = JSPUtil.getJobID(new Path(encodedLogFileName).getName());
   Format decimal = new DecimalFormat();
 
   FileSystem fs = (FileSystem) application.getAttribute("fileSys");
-  JobHistory.JobInfo job = JSPUtil.getJobInfo(request, fs);
+  JobTracker jobTracker = (JobTracker) application.getAttribute("job.tracker");
+  JobHistory.JobInfo job = JSPUtil.checkAccessAndGetJobInfo(request,
+      response, jobTracker, fs, new Path(logFile));
+  if (job == null) {
+    return;
+  }
 
-  JobHistory.Task task = job.getAllTasks().get(taskid);
+  JobHistory.Task task = job.getAllTasks().get(tipid.toString());
   JobHistory.TaskAttempt attempt = task.getTaskAttempts().get(attemptid);
 
   Counters counters = 
@@ -101,7 +109,7 @@
 %>
 
 <hr>
-<a href="jobdetailshistory.jsp?jobid=<%=jobid%>&logFile=<%=encodedLogFileName%>">Go back to the job</a><br>
+<a href="jobdetailshistory.jsp?logFile=<%=encodedLogFileName%>">Go back to the job</a><br>
 <a href="jobtracker.jsp">Go back to JobTracker</a><br>
 <%
 out.println(ServletUtil.htmlFooter());