Restrict os.read within a Task to only allow reading from upstream task PathRefs (#5037)

Fixes #5023, the last bullet
of #3746 we had punted on
earlier

With this PR:

1. A normal `Task` can only read from sub-paths of `PathRef`s returned
by the upstream tasks it depends on.
2. Like the write restrictions implemented earlier, we do not restrict
`Command`s (since those are expected to do arbitrary side-effecting
things),
3. For reads we are also lenient for `InputImpl`s and their sub-types
`SourceImpl`s and `SourcesImpl`s (since those are the roots of the build
graph and so are expected to read from arbitrary files)

To implement this, we need some way to traverse the return value of
upstream tasks. This is done by instrumenting the
`PathRef.jsonFormatter` that is called during the recursive
serialization/deserialization of task return values during cache saves
and loads. There is no other place for us to put this logic unless we
add a second typeclass that task return values would have to implement,
so this hackery piggy-backing on JSON serialization and de-serialization
will have to do for now

Added a single example test case to make sure this catches things,
relying on the bulk of the existing test suite to ensure this doesn't
break things. Also had to fix a bunch of existing violations of this
rule, mostly by adjusting the code to use the proper pattern, but for
some violations I just grandfathered them in for now. Definitely seems
like a good guardrail to have given how often we screw it up ourselves
in Mill's own repo

For now this only limits reads during the _execution_ phase, and does
not limit reads during the _resolution_ phase. That can come in a follow
up

Author: lihaoyi

contrib/scalapblib/test/src/mill/contrib/scalapblib/TutorialTests.scala

@@ -56,8 +56,8 @@ object TutorialTests extends TestSuite {
       }
 
       override def scalaPBSearchDeps = true
-      override def scalaPBIncludePath = Seq(
-        PathRef(moduleDir / "protobuf/tutorial")
+      override def scalaPBIncludePath = Task.Sources(
+        moduleDir / "protobuf/tutorial"
       )
     }
     lazy val millDiscover = Discover[this.type]

core/define/src/mill/define/PathRef.scala

@@ -82,6 +82,9 @@ object PathRef {
   private[mill] val validatedPaths: DynamicVariable[ValidatedPaths] =
     new DynamicVariable[ValidatedPaths](new ValidatedPaths())
 
+  private[mill] val serializedPaths: DynamicVariable[List[PathRef]] =
+    new DynamicVariable(null)
+
   class PathRefValidationException(val pathRef: PathRef)
       extends RuntimeException(s"Invalid path signature detected: ${pathRef}")
 
@@ -169,11 +172,29 @@ object PathRef {
     new PathRef(path, quick, sig, revalidate)
   }
 
+  private[mill] def withSerializedPaths[T](block: => T): (T, Seq[PathRef]) = {
+    serializedPaths.value = Nil
+    try {
+      val res = block
+      (res, serializedPaths.value)
+    } finally serializedPaths.value = null
+  }
+  private def storeSerializedPaths(pr: PathRef) = {
+    if (serializedPaths.value != null) {
+      serializedPaths.synchronized {
+        serializedPaths.value = pr :: serializedPaths.value
+      }
+    }
+  }
+
   /**
    * Default JSON formatter for [[PathRef]].
    */
   implicit def jsonFormatter: RW[PathRef] = upickle.default.readwriter[String].bimap[PathRef](
-    p => p.toString(),
+    p => {
+      storeSerializedPaths(p)
+      p.toString()
+    },
     s => {
       val Array(prefix, valid0, hex, pathString) = s.split(":", 4)
 
@@ -192,6 +213,7 @@ object PathRef {
       val sig = java.lang.Long.parseLong(hex, 16).toInt
       val pr = PathRef(path, quick, sig, revalidate = validOrig)
       validatedPaths.value.revalidateIfNeededOrThrow(pr)
+      storeSerializedPaths(pr)
       pr
     }
   )

core/define/src/mill/define/internal/ResolveChecker.scala

@@ -1,10 +1,10 @@
 package mill.define.internal
 
-object ResolveChecker extends os.Checker {
+class ResolveChecker(workspace: os.Path) extends os.Checker {
   def onRead(path: os.ReadablePath): Unit = ()
 
   def onWrite(path: os.Path): Unit = {
-    sys.error(s"Writing to $path not allowed during resolution phase")
+    sys.error(s"Writing to ${path.relativeTo(workspace)} not allowed during resolution phase")
   }
 
   def withResolveChecker[T](f: () => T): T = {

core/eval/src/mill/eval/EvaluatorImpl.scala

@@ -52,7 +52,7 @@ final class EvaluatorImpl private[mill] (
       allowPositionalCommandArgs: Boolean = false,
       resolveToModuleTasks: Boolean = false
   ): mill.api.Result[List[Segments]] = {
-    os.checker.withValue(ResolveChecker) {
+    os.checker.withValue(ResolveChecker(workspace)) {
       Resolve.Segments.resolve(
         rootModule,
         scriptArgs,
@@ -73,7 +73,7 @@ final class EvaluatorImpl private[mill] (
       allowPositionalCommandArgs: Boolean = false,
       resolveToModuleTasks: Boolean = false
   ): mill.api.Result[List[NamedTask[?]]] = {
-    os.checker.withValue(ResolveChecker) {
+    os.checker.withValue(ResolveChecker(workspace)) {
       Evaluator.currentEvaluator0.withValue(this) {
         Resolve.Tasks.resolve(
           rootModule,
@@ -91,7 +91,7 @@ final class EvaluatorImpl private[mill] (
       allowPositionalCommandArgs: Boolean = false,
       resolveToModuleTasks: Boolean = false
   ): mill.api.Result[List[Either[Module, NamedTask[?]]]] = {
-    os.checker.withValue(ResolveChecker) {
+    os.checker.withValue(ResolveChecker(workspace)) {
       Evaluator.currentEvaluator0.withValue(this) {
         Resolve.Inspect.resolve(
           rootModule,
@@ -249,7 +249,7 @@ final class EvaluatorImpl private[mill] (
       selectMode: SelectMode,
       selectiveExecution: Boolean = false
   ): mill.api.Result[Evaluator.Result[Any]] = {
-    val resolved = os.checker.withValue(ResolveChecker) {
+    val resolved = os.checker.withValue(ResolveChecker(workspace)) {
       Evaluator.currentEvaluator0.withValue(this) {
         Resolve.Tasks.resolve(
           rootModule,

core/exec/src/mill/exec/Execution.scala

@@ -166,7 +166,7 @@ private[mill] case class Execution(
             .toMap
 
           futures(terminal) = Future.successful(
-            Some(GroupExecution.Results(taskResults, group.toSeq, false, -1, -1, false))
+            Some(GroupExecution.Results(taskResults, group.toSeq, false, -1, -1, false, Nil))
           )
         } else {
           futures(terminal) = Future.sequence(deps.map(futures)).map { upstreamValues =>
@@ -186,6 +186,11 @@ private[mill] case class Execution(
                   .flatMap(_.iterator.flatMap(_.newResults))
                   .toMap
 
+                val upstreamPathRefs = upstreamValues
+                  .iterator
+                  .flatMap(_.iterator.flatMap(_.serializedPaths))
+                  .toSeq
+
                 val startTime = System.nanoTime() / 1000
 
                 // should we log progress?
@@ -219,7 +224,8 @@ private[mill] case class Execution(
                   allTransitiveClassMethods,
                   forkExecutionContext,
                   exclusive,
-                  offline
+                  offline,
+                  upstreamPathRefs
                 )
 
                 // Count new failures - if there are upstream failures, tasks should be skipped, not failed