<commit_before>package at.hazm.webserver.handler

import java.io._
import java.nio.charset.StandardCharsets
import java.nio.file.Path

import at.hazm._File
import at.hazm.webserver.Server
import at.hazm.webserver.handler.CommandHandler._
import com.twitter.finagle.http._
import com.twitter.io.{Bufs, Reader}
import org.slf4j.{Logger, LoggerFactory}

import scala.annotation.tailrec
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.matching.Regex
import scala.util.{Failure, Success, Try}

/**
  * docroot 上の特定のディレクトリに配置されているコマンドをサーバサイドで実行するためのリクエストハンドラです。
  *
  * @param docroot      ドキュメントルート
  * @param timeout      実行タイムアウト (ミリ秒)
  * @param prefix       URL 上のプレフィクス
  * @param interpreters (`.js` などの拡張子, インタープリタ) のマップ
  */
class CommandHandler(docroot:Path, timeout:Long, prefix:String, interpreters:Map[String, String]) extends RequestHandler(docroot) {
  logger.info(s"""external script enabled for prefix "$prefix", extensions [${interpreters.keys.mkString("\"", "\", \"", "\"")}]""")

  override def apply(request:Request):Option[Response] = None

  override def applyAsync(request:Request)(implicit context:ExecutionContext):Future[Option[Response]] = if(request.uri.startsWith(prefix)) {
    FileHandler.mapLocalFile(docroot, request.uri).map { file =>
      interpreters.get("." + file.getExtension) match {
        case Some(interpreter) =>
          if(!file.isFile) {
            Future.successful(Some(getErrorResponse(request, Status.NotFound)))
          } else {
            startup(request, interpreter, file).map(x => Some(x))
          }
        case None =>
          Future.successful(Some(getErrorResponse(request, Status.Forbidden)))
      }
    }.getOrElse {
      Future.successful(Some(getErrorResponse(request, Status.BadRequest)))
    }
  } else Future.successful(None)

  /**
    * リクエストに応答するために指定されたコマンドを実行しレスポンスを作成します。
    *
    * @param request     リクエスト
    * @param interpreter 起動するインタープリタ
    * @param file        実行するファイル
    * @return レスポンス
    */
  private[this] def startup(request:Request, interpreter:String, file:File)(implicit context:ExecutionContext):Future[Response] = {
    val pipeRequestBody = request.method == Method.Post || request.method == Method.Put

    // プロセスの構築
    val builder = new ProcessBuilder()
      .command(interpreter, file.getName)
      .directory(file.getParentFile)
      .redirectError(ProcessBuilder.Redirect.PIPE)
      .redirectInput(ProcessBuilder.Redirect.PIPE)
      .redirectOutput(ProcessBuilder.Redirect.PIPE)

    // CGI と同様の環境変数を設定
    val env = builder.environment()
    env.put("DOCUMENT_ROOT", docroot.toString)
    env.put("GATEWAY_INTERFACE", "CGI/1.1")
    request.headerMap.keys.filter { key =>
      !key.equalsIgnoreCase("Content-Length") && !key.equalsIgnoreCase("Content-Type")
    }.foreach { key =>
      val e = key.toUpperCase.replace('-', '_').replaceAll("[\\x00-\\x1F\\x7F]", "_")
      env.put("HTTP_" + e, request.headerMap.get(key).getOrElse(""))
    }
    if(request.method == Method.Get) {
      val uri = request.uri
      val sep = uri.indexOf('?')
      if(sep >= 0) {
        env.put("QUERY_STRING", uri.substring(sep + 1))
      }
    } else if(pipeRequestBody) {
      env.put("CONTENT_LENGTH", request.contentLength.getOrElse(0L).toString)
      env.put("CONTENT_TYPE", request.contentType.getOrElse(""))
    }
    env.put("REMOTE_ADDR", request.remoteAddress.getHostAddress)
    env.put("REMOTE_PORT", request.remotePort.toString)
    env.put("REQUEST_METHOD", request.method.name)
    env.put("REQUEST_URI", request.uri)
    env.put("SCRIPT_FILENAME", file.toString)
    env.put("SCRIPT_NAME", request.path)

    val start = System.currentTimeMillis()

    def limitTime = timeout - (System.currentTimeMillis() - start)

    def error(status:Status) = getErrorResponse(request, status)

    case class GatewayTimeout() extends Exception

    class ErrorReader(name:String, is:InputStream) extends Thread {
      override def run():Unit = try {
        val in = new BufferedReader(new InputStreamReader(is))
        Iterator.continually(in.readLine()).takeWhile(_ != null).foreach(line => logger.warn(s"[$name] $line"))
      } catch {
        case ex:Throwable =>
          logger.warn(s"while reading stderr of process: $name", ex)
      }
    }

    Server.scheduler.runWithTimeout(limitTime, { _ =>
      Future.failed(GatewayTimeout())
    }) {
      Try(builder.start())
    }.flatMap {
      case Failure(_:GatewayTimeout) =>
        logger.warn(f"${request.path}: ${System.currentTimeMillis() - start}%,dms => start timeout")
        Future.successful(error(Status.GatewayTimeout))
      case Failure(ex) =>
        logger.error(f"${request.path}: ${System.currentTimeMillis() - start}%,dms => error", ex)
        Future.successful(error(Status.InternalServerError))
      case Success(proc) =>
        // コマンド実行
        val promise = Promise[Response]()
        Server.scheduler.runWithTimeout(limitTime, { _ =>
          logger.warn(f"${request.path}: ${System.currentTimeMillis() - start}%,dms => execution timeout")
          promise.success(error(Status.GatewayTimeout))
          proc.destroyForcibly()
          Future.unit
        }) {
          try {
            new ErrorReader(interpreter, proc.getErrorStream).start()

            // リクエストボディを標準入力経由でコマンドに渡す
            if(pipeRequestBody) {
              val out = proc.getOutputStream
              out.write(Bufs.ownedByteArray(request.content))
              out.flush()
            }
            val response = pipeResponse(request, proc.getInputStream)
            if(!promise.isCompleted) {
              if(logger.isDebugEnabled) {
                val tm = System.currentTimeMillis() - start
                logger.debug(f"${request.path}%s: $tm%,dms => ${response.status.code} ${response.status.reason}")
              }
              promise.success(response)
            }
          } catch {
            case ex:Throwable =>
              logger.error(s"fail to execute external command: $file", ex)
              if(!promise.isCompleted) {
                promise.success(error(Status.InternalServerError))
              }
          }
        }
        promise.future
    }
  }

  /**
    * 指定された入力ストリームから外部スクリプトのレスポンスを読み込んでレスポンスを構築します。
    *
    * @param request エラーメッセージ用のリクエスト
    * @param is      外部スクリプトの出力
    * @return レスポンス
    */
  private[this] def pipeResponse(request:Request, is:InputStream):Response = {
    val in = new PushbackInputStream(is)
    readHeader(in).flatMap { case (version, status, headers) =>
      Try {
        val reader = Reader.fromStream(in)
        val response = Response(version, status, reader)
        headers.foreach { case (name, value) =>
          response.headerMap.add(name, value)
        }
        response
      }.recoverWith { case ex =>
        logger.error(s"invalid response header from script: $version ${status.code} ${headers.seq.map(x => x._1 + ":" + x._2).mkString("[", ", ", "]")}; ${request.uri}", ex)
        Failure(ex)
      }.toOption
    }.getOrElse(getErrorResponse(request, Status.InternalServerError))
  }

}

private object CommandHandler {
  val logger:Logger = LoggerFactory.getLogger(classOf[CommandHandler])

  /** HTTP レスポンスのステータス行と一致する正規表現 */
  private[this] val statusLine:Regex =
    """HTTP/(\d+)\.(\d+)\s+(\d+)\s+(.*)""".r

  /**
    * 指定された入力ストリームからステータス行とヘッダを読み込みます。読み込みが終わる前に EOF を検出した場合は None を返します。
    *
    * @param in 入力ストリーム
    * @return ステータス行とヘッダ<commit_msg>change to ignore invalid header field from command-handler<commit_after>package at.hazm.webserver.handler

import java.io._
import java.nio.charset.StandardCharsets
import java.nio.file.Path

import at.hazm._File
import at.hazm.webserver.Server
import at.hazm.webserver.handler.CommandHandler._
import com.twitter.finagle.http._
import com.twitter.io.{Bufs, Reader}
import org.slf4j.{Logger, LoggerFactory}

import scala.annotation.tailrec
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.matching.Regex
import scala.util.{Failure, Success, Try}

/**
  * docroot 上の特定のディレクトリに配置されているコマンドをサーバサイドで実行するためのリクエストハンドラです。
  *
  * @param docroot      ドキュメントルート
  * @param timeout      実行タイムアウト (ミリ秒)
  * @param prefix       URL 上のプレフィクス
  * @param interpreters (`.js` などの拡張子, インタープリタ) のマップ
  */
class CommandHandler(docroot:Path, timeout:Long, prefix:String, interpreters:Map[String, String]) extends RequestHandler(docroot) {
  logger.info(s"""external script enabled for prefix "$prefix", extensions [${interpreters.keys.mkString("\"", "\", \"", "\"")}]""")

  override def apply(request:Request):Option[Response] = None

  override def applyAsync(request:Request)(implicit context:ExecutionContext):Future[Option[Response]] = if(request.uri.startsWith(prefix)) {
    FileHandler.mapLocalFile(docroot, request.uri).map { file =>
      interpreters.get("." + file.getExtension) match {
        case Some(interpreter) =>
          if(!file.isFile) {
            Future.successful(Some(getErrorResponse(request, Status.NotFound)))
          } else {
            startup(request, interpreter, file).map(x => Some(x))
          }
        case None =>
          Future.successful(Some(getErrorResponse(request, Status.Forbidden)))
      }
    }.getOrElse {
      Future.successful(Some(getErrorResponse(request, Status.BadRequest)))
    }
  } else Future.successful(None)

  /**
    * リクエストに応答するために指定されたコマンドを実行しレスポンスを作成します。
    *
    * @param request     リクエスト
    * @param interpreter 起動するインタープリタ
    * @param file        実行するファイル
    * @return レスポンス
    */
  private[this] def startup(request:Request, interpreter:String, file:File)(implicit context:ExecutionContext):Future[Response] = {
    val pipeRequestBody = request.method == Method.Post || request.method == Method.Put

    // プロセスの構築
    val builder = new ProcessBuilder()
      .command(interpreter, file.getName)
      .directory(file.getParentFile)
      .redirectError(ProcessBuilder.Redirect.PIPE)
      .redirectInput(ProcessBuilder.Redirect.PIPE)
      .redirectOutput(ProcessBuilder.Redirect.PIPE)

    // CGI と同様の環境変数を設定
    val env = builder.environment()
    env.put("DOCUMENT_ROOT", docroot.toString)
    env.put("GATEWAY_INTERFACE", "CGI/1.1")
    request.headerMap.keys.filter { key =>
      !key.equalsIgnoreCase("Content-Length") && !key.equalsIgnoreCase("Content-Type")
    }.foreach { key =>
      val e = key.toUpperCase.replace('-', '_').replaceAll("[\\x00-\\x1F\\x7F]", "_")
      env.put("HTTP_" + e, request.headerMap.get(key).getOrElse(""))
    }
    if(request.method == Method.Get) {
      val uri = request.uri
      val sep = uri.indexOf('?')
      if(sep >= 0) {
        env.put("QUERY_STRING", uri.substring(sep + 1))
      }
    } else if(pipeRequestBody) {
      env.put("CONTENT_LENGTH", request.contentLength.getOrElse(0L).toString)
      env.put("CONTENT_TYPE", request.contentType.getOrElse(""))
    }
    env.put("REMOTE_ADDR", request.remoteAddress.getHostAddress)
    env.put("REMOTE_PORT", request.remotePort.toString)
    env.put("REQUEST_METHOD", request.method.name)
    env.put("REQUEST_URI", request.uri)
    env.put("SCRIPT_FILENAME", file.toString)
    env.put("SCRIPT_NAME", request.path)

    val start = System.currentTimeMillis()

    def limitTime = timeout - (System.currentTimeMillis() - start)

    def error(status:Status) = getErrorResponse(request, status)

    case class GatewayTimeout() extends Exception

    class ErrorReader(name:String, is:InputStream) extends Thread {
      override def run():Unit = try {
        val in = new BufferedReader(new InputStreamReader(is))
        Iterator.continually(in.readLine()).takeWhile(_ != null).foreach(line => logger.warn(s"[$name] $line"))
      } catch {
        case ex:Throwable =>
          logger.warn(s"while reading stderr of process: $name", ex)
      }
    }

    Server.scheduler.runWithTimeout(limitTime, { _ =>
      Future.failed(GatewayTimeout())
    }) {
      Try(builder.start())
    }.flatMap {
      case Failure(_:GatewayTimeout) =>
        logger.warn(f"${request.path}: ${System.currentTimeMillis() - start}%,dms => start timeout")
        Future.successful(error(Status.GatewayTimeout))
      case Failure(ex) =>
        logger.error(f"${request.path}: ${System.currentTimeMillis() - start}%,dms => error", ex)
        Future.successful(error(Status.InternalServerError))
      case Success(proc) =>
        // コマンド実行
        val promise = Promise[Response]()
        Server.scheduler.runWithTimeout(limitTime, { _ =>
          logger.warn(f"${request.path}: ${System.currentTimeMillis() - start}%,dms => execution timeout")
          promise.success(error(Status.GatewayTimeout))
          proc.destroyForcibly()
          Future.unit
        }) {
          try {
            new ErrorReader(interpreter, proc.getErrorStream).start()

            // リクエストボディを標準入力経由でコマンドに渡す
            if(pipeRequestBody) {
              val out = proc.getOutputStream
              out.write(Bufs.ownedByteArray(request.content))
              out.flush()
            }
            val response = pipeResponse(request, proc.getInputStream)
            if(!promise.isCompleted) {
              if(logger.isDebugEnabled) {
                val tm = System.currentTimeMillis() - start
                logger.debug(f"${request.path}%s: $tm%,dms => ${response.status.code} ${response.status.reason}")
              }
              promise.success(response)
            }
          } catch {
            case ex:Throwable =>
              logger.error(s"fail to execute external command: $file", ex)
              if(!promise.isCompleted) {
                promise.success(error(Status.InternalServerError))
              }
          }
        }
        promise.future
    }
  }

  /**
    * 指定された入力ストリームから外部スクリプトのレスポンスを読み込んでレスポンスを構築します。
    *
    * @param request エラーメッセージ用のリクエスト
    * @param is      外部スクリプトの出力
    * @return レスポンス
    */
  private[this] def pipeResponse(request:Request, is:InputStream):Response = {
    val in = new PushbackInputStream(is)
    readHeader(in).flatMap { case (version, status, headers) =>
      Try {
        val reader = Reader.fromStream(in)
        val response = Response(version, status, reader)
        headers.foreach { case (name, value) =>
          try {
            response.headerMap.add(name, value)
          } catch {
            case ex:IllegalArgumentException =>
              logger.warn(s"the header generated by command-handler contained an invalid field: $name:$value (ignored); $ex")
          }
        }
        response
      }.recoverWith { case ex =>
        logger.error(s"invalid response header from script: $version ${status.code} ${headers.seq.map(x => x._1 + ":" + x._2).mkString("[", ", ", "]")}; ${request.uri}", ex)
        Failure(ex)
      }.toOption
    }.getOrElse(getErrorResponse(request, Status.InternalServerError))
  }

}

private object CommandHandler {
  val logger:Logger = LoggerFactory.getLogger(classOf[CommandHandler])

  /** HTTP レスポンスのステータス行と一致する正規表現 */
  private[this] val statusLine:Regex =
    """HTTP/(\d+)\.(\d+)\s+(\d+)\s+(.*)""".r

  /**
    * 指定された入力ストリームからステータス行とヘッダを読み込みます。読み込みが終わる前に EOF を検出した場合は None を返します。
    *
    * @param in 入力ストリーム
    * @return ステータス行とヘッダ
{'fraction_non_alphanumeric': 0.13511916486113848, 'fraction_numerical': 0.0015757337010045302, 'mean_word_length': 2.9502074688796682, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 50, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '530864', 'n_tokens_mistral': 5142, 'n_tokens_neox': 5077, 'n_words': 800}