#!/usr/bin/env python
"""Test client vfs."""

import functools
import os

from absl import app

from grr_response_client import vfs
from grr_response_client.client_actions import searching
from grr_response_core.lib import utils
from grr_response_core.lib.rdfvalues import client_fs as rdf_client_fs
from grr_response_core.lib.rdfvalues import paths as rdf_paths
from grr_response_core.lib.util import temp
from grr.test_lib import client_test_lib
from grr.test_lib import filesystem_test_lib
from grr.test_lib import test_lib
from grr.test_lib import vfs_test_lib


class MockVFSHandlerFind(vfs.VFSHandler):
  """A mock VFS handler for finding files.

  This is used to create the /mock2/ client vfs branch which is utilized in the
  below tests.
  """
  supported_pathtype = rdf_paths.PathSpec.PathType.OS

  filesystem = {
      "/": ["mock2"],
      "/mock2": ["directory1", "directory3"],
      "/mock2/directory1": ["file1.txt", "file2.txt", "directory2"],
      "/mock2/directory1/file1.txt": b"Secret 1",
      "/mock2/directory1/file2.txt": b"Another file",
      "/mock2/directory1/directory2": ["file.jpg", "file.mp3"],
      "/mock2/directory1/directory2/file.jpg": b"JPEG",
      "/mock2/directory1/directory2/file.mp3": b"MP3 movie",
      "/mock2/directory3": ["file1.txt", "long_file.text"],
      "/mock2/directory3/file1.txt": b"A text file",
      "/mock2/directory3/long_file.text": (b"space " * 100000 + b"A Secret")
  }

  def __init__(self, base_fd, handlers, pathspec=None, progress_callback=None):
    super().__init__(
        base_fd,
        handlers=handlers,
        pathspec=pathspec,
        progress_callback=progress_callback)

    self.pathspec.Append(pathspec)
    self.path = self.pathspec.CollapsePath()

    try:
      self.content = self.filesystem[self.path]
      if isinstance(self.content, bytes):
        self.size = len(self.content)
    except KeyError:
      raise IOError("not mocking %s" % self.path)

  def Read(self, length):
    # Reading the mocked directory raises.
    if isinstance(self.content, list):
      raise IOError()

    result = self.content[self.offset:self.offset + length]
    self.offset = min(self.size, self.offset + len(result))
    return result

  def ListNames(self):
    return self.content

  def DoStat(self, path):
    result = rdf_client_fs.StatEntry()
    if path.startswith("/mock2/directory3"):
      result.st_dev = 1
    else:
      result.st_dev = 2
    f = self.filesystem[path]
    if isinstance(f, bytes):
      if path.startswith("/mock2/directory1/directory2"):
        result.st_mode = 0o0100644  # u=rw,g=r,o=r on regular file
        result.st_uid = 50
        result.st_gid = 500
      elif path.startswith("/mock2/directory3"):
        result.st_mode = 0o0100643  # u=rw,g=r,o=wx on regular file
        result.st_uid = 60
        result.st_gid = 600
      else:
        result.st_mode = 0o0104666  # setuid, u=rw,g=rw,o=rw on regular file
        result.st_uid = 90
        result.st_gid = 900
    else:
      result.st_mode = 0o0040775  # u=rwx,g=rwx,o=rx on directory
      result.st_uid = 0
      result.st_gid = 4
    result.st_size = len(f)
    result.st_mtime = 1373185602

    return result

  def ListFiles(self, ext_attrs=None):
    """Mock the filesystem."""
    del ext_attrs  # Unused.

    for child in self.content:
      # We have a mock FS here that only uses "/".
      path = "/".join([self.path, child])
      result = self.DoStat(path)
      ps = self.pathspec.Copy()
      ps.Append(path=child, pathtype=self.supported_pathtype)
      result.pathspec = ps
      yield result

  def IsDirectory(self):
    return bool(self.content)

  def Stat(
      self,
      ext_attrs: bool = False,
      follow_symlink: bool = True,
  ) -> rdf_client_fs.StatEntry:
    del ext_attrs, follow_symlink  # Unused.

    result = self.DoStat(self.path)
    result.pathspec = self.pathspec
    return result


def SearchParams(block_size, envelope_size):

  def Decorator(func):

    @functools.wraps(func)
    def _SearchParams(*args, **kwargs):
      """Wrapper function that sets and restores search parameters."""

      old_sizes = (searching.Grep.BUFF_SIZE, searching.Grep.ENVELOPE_SIZE)
      searching.Grep.BUFF_SIZE = block_size
      searching.Grep.ENVELOPE_SIZE = envelope_size
      try:
        return func(*args, **kwargs)
      finally:
        searching.Grep.BUFF_SIZE, searching.Grep.ENVELOPE_SIZE = old_sizes

    return _SearchParams

  return Decorator


class FindTest(client_test_lib.EmptyActionTest):
  """Test the find client Actions."""

  def setUp(self):
    super().setUp()

    # Install the mock
    vfs_overrider = vfs_test_lib.VFSOverrider(rdf_paths.PathSpec.PathType.OS,
                                              MockVFSHandlerFind)
    vfs_overrider.Start()
    self.addCleanup(vfs_overrider.Stop)

  def testFindAction(self):
    """Test the find action."""
    # First get all the files at once
    pathspec = rdf_paths.PathSpec(
        path="/mock2/", pathtype=rdf_paths.PathSpec.PathType.OS)
    request = rdf_client_fs.FindSpec(pathspec=pathspec, path_regex=".")
    result = self.RunAction(searching.Find, request)
    self.assertCountEqual([r.pathspec.Basename() for r in result], [
        "file1.txt", "file2.txt", "file.jpg", "file.mp3", "directory2",
        "directory1", "directory3"
    ])

  def testFindAction2(self):
    """Test the find action path regex."""
    pathspec = rdf_paths.PathSpec(
        path="/mock2/", pathtype=rdf_paths.PathSpec.PathType.OS)
    request = rdf_client_fs.FindSpec(pathspec=pathspec, path_regex=".*mp3")
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 1)
    self.assertEqual(all_files[0].pathspec.Basename(), "file.mp3")

  def testFindAction3(self):
    """Test the find action data regex."""
    # First get all the files at once
    pathspec = rdf_paths.PathSpec(
        path="/mock2/", pathtype=rdf_paths.PathSpec.PathType.OS)
    request = rdf_client_fs.FindSpec(
        pathspec=pathspec, data_regex=b"Secret", cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 2)
    self.assertEqual(all_files[0].pathspec.Basename(), "file1.txt")
    self.assertEqual(all_files[1].pathspec.Basename(), "long_file.text")

  def testFindSizeLimits(self):
    """Test the find action size limits."""
    # First get all the files at once
    request = rdf_client_fs.FindSpec(
        min_file_size=4, max_file_size=15, cross_devs=True)
    request.pathspec.Append(
        path="/mock2/", pathtype=rdf_paths.PathSpec.PathType.OS)

    results = self.RunAction(searching.Find, request)
    all_files = [r.pathspec.Basename() for r in results]
    self.assertLen(all_files, 5)

    for filename in all_files:
      # Our mock filesize is the length of the base filename, check all the
      # files we got match the size criteria
      self.assertBetween(len(filename), 4, 15)

  def testNoFilters(self):
    """Test the we get all files with no filters in place."""
    # First get all the files at once
    pathspec = rdf_paths.PathSpec(
        path="/mock2/", pathtype=rdf_paths.PathSpec.PathType.OS)
    request = rdf_client_fs.FindSpec(pathspec=pathspec, cross_devs=True)
    all_files = self.RunAction(searching.Find, request)
    self.assertLen(all_files, 9)

  def testFindActionCrossDev(self):
    """Test that devices boundaries don't get crossed, also by default."""
    pathspec = rdf_paths.PathSpec(
        path="/mock2/", pathtype=rdf_paths.PathSpec.PathType.OS)
    request = rdf_client_fs.FindSpec(
        pathspec=pathspec, cross_devs=True, path_regex=".")
    all_files = self.RunAction(searching.Find, request)
    self.assertLen(all_files, 9)

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec, cross_devs=False, path_regex=".")
    all_files = self.RunAction(searching.Find, request)
    self.assertLen(all_files, 7)

    request = rdf_client_fs.FindSpec(pathspec=pathspec, path_regex=".")
    all_files = self.RunAction(searching.Find, request)
    self.assertLen(all_files, 7)

  def testPermissionFilter(self):
    """Test filtering based on file/folder permission happens correctly."""

    pathspec = rdf_paths.PathSpec(
        path="/mock2/", pathtype=rdf_paths.PathSpec.PathType.OS)

    # Look for files that match exact permissions

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec, path_regex=".", perm_mode=0o644, cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 2)
    self.assertEqual(all_files[0].pathspec.Dirname().Basename(), "directory2")
    self.assertEqual(all_files[0].pathspec.Basename(), "file.jpg")
    self.assertEqual(all_files[1].pathspec.Dirname().Basename(), "directory2")
    self.assertEqual(all_files[1].pathspec.Basename(), "file.mp3")

    # Look for files/folders where 'others' have 'write' permission. All other
    # attributes don't matter. Setuid bit must also be set and guid or sticky
    # bit must not be set.

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec,
        path_regex=".",
        perm_mode=0o4002,
        perm_mask=0o7002,
        cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 2)
    self.assertEqual(all_files[0].pathspec.Dirname().Basename(), "directory1")
    self.assertEqual(all_files[0].pathspec.Basename(), "file1.txt")
    self.assertEqual(all_files[1].pathspec.Dirname().Basename(), "directory1")
    self.assertEqual(all_files[1].pathspec.Basename(), "file2.txt")

    # Look for files where 'others' have 'execute' permission. All other
    # attributes don't matter. Only look for 'regular' files.

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec,
        path_regex=".",
        perm_mode=0o0100001,
        perm_mask=0o0100001,
        cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 2)
    self.assertEqual(all_files[0].pathspec.Dirname().Basename(), "directory3")
    self.assertEqual(all_files[0].pathspec.Basename(), "file1.txt")
    self.assertEqual(all_files[1].pathspec.Dirname().Basename(), "directory3")
    self.assertEqual(all_files[1].pathspec.Basename(), "long_file.text")

    # Look for folders where 'group' have 'execute' permission. All other
    # attributes don't matter. Only look for folders.

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec,
        path_regex=".",
        perm_mode=0o0040010,
        perm_mask=0o0040010,
        cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 3)
    self.assertEqual(all_files[0].pathspec.Basename(), "directory2")
    self.assertEqual(all_files[1].pathspec.Basename(), "directory1")
    self.assertEqual(all_files[2].pathspec.Basename(), "directory3")

  def testUIDFilter(self):
    """Test filtering based on uid happens correctly."""

    pathspec = rdf_paths.PathSpec(
        path="/mock2/", pathtype=rdf_paths.PathSpec.PathType.OS)

    # Look for files that have uid of 60

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec, path_regex=".", uid=60, cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 2)
    self.assertEqual(all_files[0].pathspec.Dirname().Basename(), "directory3")
    self.assertEqual(all_files[0].pathspec.Basename(), "file1.txt")
    self.assertEqual(all_files[1].pathspec.Dirname().Basename(), "directory3")
    self.assertEqual(all_files[1].pathspec.Basename(), "long_file.text")

    # Look for files that have uid of 0

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec, path_regex=".", uid=0, cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 3)
    self.assertEqual(all_files[0].pathspec.Basename(), "directory2")
    self.assertEqual(all_files[1].pathspec.Basename(), "directory1")
    self.assertEqual(all_files[2].pathspec.Basename(), "directory3")

  def testGIDFilter(self):
    """Test filtering based on gid happens correctly."""

    pathspec = rdf_paths.PathSpec(
        path="/mock2/", pathtype=rdf_paths.PathSpec.PathType.OS)

    # Look for files that have gid of 500

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec, path_regex=".", gid=500, cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 2)
    self.assertEqual(all_files[0].pathspec.Dirname().Basename(), "directory2")
    self.assertEqual(all_files[0].pathspec.Basename(), "file.jpg")
    self.assertEqual(all_files[1].pathspec.Dirname().Basename(), "directory2")
    self.assertEqual(all_files[1].pathspec.Basename(), "file.mp3")

    # Look for files that have uid of 900

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec, path_regex=".", gid=900, cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 2)
    self.assertEqual(all_files[0].pathspec.Dirname().Basename(), "directory1")
    self.assertEqual(all_files[0].pathspec.Basename(), "file1.txt")
    self.assertEqual(all_files[1].pathspec.Dirname().Basename(), "directory1")
    self.assertEqual(all_files[1].pathspec.Basename(), "file2.txt")

  def testUIDAndGIDFilter(self):
    """Test filtering based on combination of uid and gid happens correctly."""

    pathspec = rdf_paths.PathSpec(
        path="/mock2/", pathtype=rdf_paths.PathSpec.PathType.OS)

    # Look for files that have uid of 90 and gid of 500

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec, path_regex=".", uid=90, gid=500, cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertEmpty(all_files)

    # Look for files that have uid of 50 and gid of 500

    request = rdf_client_fs.FindSpec(
        pathspec=pathspec, path_regex=".", uid=50, gid=500, cross_devs=True)
    all_files = self.RunAction(searching.Find, request)

    self.assertLen(all_files, 2)
    self.assertEqual(all_files[0].pathspec.Dirname().Basename(), "directory2")
    self.assertEqual(all_files[0].pathspec.Basename(), "file.jpg")
    self.assertEqual(all_files[1].pathspec.Dirname().Basename(), "directory2")
    self.assertEqual(all_files[1].pathspec.Basename(), "file.mp3")


class FindExtAttrsTest(client_test_lib.EmptyActionTest):

  def testExtAttrsCollection(self):
    with temp.AutoTempDirPath(remove_non_empty=True) as temp_dirpath:
      foo_filepath = temp.TempFilePath(dir=temp_dirpath)
      filesystem_test_lib.SetExtAttr(
          foo_filepath, name="user.quux", value="foo")

      bar_filepath = temp.TempFilePath(dir=temp_dirpath)
      filesystem_test_lib.SetExtAttr(
          bar_filepath, name="user.quux", value="bar")

      baz_filepath = temp.TempFilePath(dir=temp_dirpath)
      filesystem_test_lib.SetExtAttr(
          baz_filepath, name="user.quux", value="baz")

      request = rdf_client_fs.FindSpec(
          pathspec=rdf_paths.PathSpec(
              path=temp_dirpath, pathtype=rdf_paths.PathSpec.PathType.OS),
          path_glob="*",
          collect_ext_attrs=True)

      hits = self.RunAction(searching.Find, request)

      self.assertLen(hits, 3)

      values = []
      for hit in hits:
        self.assertLen(hit.ext_attrs, 1)
        values.append(hit.ext_attrs[0].value)

      self.assertCountEqual(values, [b"foo", b"bar", b"baz"])


class GrepTest(vfs_test_lib.VfsTestCase, client_test_lib.EmptyActionTest):
  """Test the find client Actions."""

  XOR_IN_KEY = 0
  XOR_OUT_KEY = 0

  def setUp(self):
    super().setUp()

    # Install the mock
    vfs_overrider = vfs_test_lib.VFSOverrider(rdf_paths.PathSpec.PathType.OS,
                                              MockVFSHandlerFind)
    vfs_overrider.Start()
    self.addCleanup(vfs_overrider.Stop)
    self.filename = "/mock2/directory1/grepfile.txt"

  def testGrep(self):
    # Use the real file system.
    vfs.Init()

    request = rdf_client_fs.GrepSpec(
        literal=utils.Xor(b"10", self.XOR_IN_KEY),
        xor_in_key=self.XOR_IN_KEY,
        xor_out_key=self.XOR_OUT_KEY)
    request.target.path = os.path.join(self.base_path, "numbers.txt")
    request.target.pathtype = rdf_paths.PathSpec.PathType.OS
    request.start_offset = 0

    result = self.RunAction(searching.Grep, request)
    hits = [x.offset for x in result]
    self.assertEqual(hits, [
        18, 288, 292, 296, 300, 304, 308, 312, 316, 320, 324, 329, 729, 1129,
        1529, 1929, 2329, 2729, 3129, 3529, 3888
    ])
    for x in result:
      self.assertIn(b"10", utils.Xor(x.data, self.XOR_OUT_KEY))
      self.assertEqual(request.target.path, x.pathspec.path)

  def testGrepRegex(self):
    # Use the real file system.
    vfs.Init()

    request = rdf_client_fs.GrepSpec(
        regex=b"1[0]",
        xor_out_key=self.XOR_OUT_KEY,
        start_offset=0,
        target=rdf_paths.PathSpec(
            path=os.path.join(self.base_path, "numbers.txt"),
            pathtype=rdf_paths.PathSpec.PathType.OS))

    result = self.RunAction(searching.Grep, request)
    hits = [x.offset for x in result]
    self.assertEqual(hits, [
        18, 288, 292, 296, 300, 304, 308, 312, 316, 320, 324, 329, 729, 1129,
        1529, 1929, 2329, 2729, 3129, 3529, 3888
    ])
    for x in result:
      self.assertIn(b"10", utils.Xor(x.data, self.XOR_OUT_KEY))

  def testGrepLength(self):
    data = b"X" * 100 + b"HIT"

    MockVFSHandlerFind.filesystem[self.filename] = data

    request = rdf_client_fs.GrepSpec(
        literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
        xor_in_key=self.XOR_IN_KEY,
        xor_out_key=self.XOR_OUT_KEY)
    request.target.path = self.filename
    request.target.pathtype = rdf_paths.PathSpec.PathType.OS
    request.start_offset = 0

    result = self.RunAction(searching.Grep, request)
    self.assertLen(result, 1)
    self.assertEqual(result[0].offset, 100)

    request = rdf_client_fs.GrepSpec(
        literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
        xor_in_key=self.XOR_IN_KEY,
        xor_out_key=self.XOR_OUT_KEY)
    request.target.path = self.filename
    request.target.pathtype = rdf_paths.PathSpec.PathType.OS
    request.start_offset = 0
    request.length = 100

    result = self.RunAction(searching.Grep, request)
    self.assertEmpty(result)

  def testGrepOffset(self):
    data = b"X" * 10 + b"HIT" + b"X" * 100

    MockVFSHandlerFind.filesystem[self.filename] = data

    request = rdf_client_fs.GrepSpec(
        literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
        xor_in_key=self.XOR_IN_KEY,
        xor_out_key=self.XOR_OUT_KEY)
    request.target.path = self.filename
    request.target.pathtype = rdf_paths.PathSpec.PathType.OS
    request.start_offset = 0

    result = self.RunAction(searching.Grep, request)
    self.assertLen(result, 1)
    self.assertEqual(result[0].offset, 10)

    request = rdf_client_fs.GrepSpec(
        literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
        xor_in_key=self.XOR_IN_KEY,
        xor_out_key=self.XOR_OUT_KEY)
    request.target.path = self.filename
    request.target.pathtype = rdf_paths.PathSpec.PathType.OS
    request.start_offset = 5

    result = self.RunAction(searching.Grep, request)
    self.assertLen(result, 1)
    # This should still report 10.
    self.assertEqual(result[0].offset, 10)

    request = rdf_client_fs.GrepSpec(
        literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
        xor_in_key=self.XOR_IN_KEY,
        xor_out_key=self.XOR_OUT_KEY)
    request.target.path = self.filename
    request.target.pathtype = rdf_paths.PathSpec.PathType.OS
    request.start_offset = 11

    result = self.RunAction(searching.Grep, request)
    self.assertEmpty(result)

  def testOffsetAndLength(self):

    data = b"X" * 10 + b"HIT" + b"X" * 100 + b"HIT" + b"X" * 10
    MockVFSHandlerFind.filesystem[self.filename] = data

    request = rdf_client_fs.GrepSpec(
        literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
        xor_in_key=self.XOR_IN_KEY,
        xor_out_key=self.XOR_OUT_KEY)
    request.target.path = self.filename
    request.target.pathtype = rdf_paths.PathSpec.PathType.OS
    request.start_offset = 11
    request.length = 100

    result = self.RunAction(searching.Grep, request)
    self.assertEmpty(result)

  @SearchParams(1000, 100)
  def testSecondBuffer(self):

    data = b"X" * 1500 + b"HIT" + b"X" * 100
    MockVFSHandlerFind.filesystem[self.filename] = data

    request = rdf_client_fs.GrepSpec(
        literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
        xor_in_key=self.XOR_IN_KEY,
        xor_out_key=self.XOR_OUT_KEY)
    request.target.path = self.filename
    request.target.pathtype = rdf_paths.PathSpec.PathType.OS
    request.start_offset = 0

    result = self.RunAction(searching.Grep, request)
    self.assertLen(result, 1)
    self.assertEqual(result[0].offset, 1500)

  @SearchParams(1000, 100)
  def testBufferBoundaries(self):

    for offset in range(-20, 20):

      data = b"X" * (1000 + offset) + b"HIT" + b"X" * 100
      MockVFSHandlerFind.filesystem[self.filename] = data

      request = rdf_client_fs.GrepSpec(
          literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
          xor_in_key=self.XOR_IN_KEY,
          xor_out_key=self.XOR_OUT_KEY)
      request.target.path = self.filename
      request.target.pathtype = rdf_paths.PathSpec.PathType.OS
      request.start_offset = 0

      result = self.RunAction(searching.Grep, request)
      self.assertLen(result, 1)
      self.assertEqual(result[0].offset, 1000 + offset)
      expected = b"X" * 10 + b"HIT" + b"X" * 10
      self.assertLen(expected, result[0].length)
      self.assertEqual(utils.Xor(result[0].data, self.XOR_OUT_KEY), expected)

  def testSnippetSize(self):

    data = b"X" * 100 + b"HIT" + b"X" * 100
    MockVFSHandlerFind.filesystem[self.filename] = data

    for before in [50, 10, 1, 0]:
      for after in [50, 10, 1, 0]:
        request = rdf_client_fs.GrepSpec(
            literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
            xor_in_key=self.XOR_IN_KEY,
            xor_out_key=self.XOR_OUT_KEY)
        request.target.path = self.filename
        request.target.pathtype = rdf_paths.PathSpec.PathType.OS
        request.start_offset = 0
        request.bytes_before = before
        request.bytes_after = after

        result = self.RunAction(searching.Grep, request)
        self.assertLen(result, 1)
        self.assertEqual(result[0].offset, 100)
        expected = b"X" * before + b"HIT" + b"X" * after
        self.assertLen(expected, result[0].length)
        self.assertEqual(utils.Xor(result[0].data, self.XOR_OUT_KEY), expected)

  @SearchParams(100, 50)
  def testGrepEverywhere(self):

    for offset in range(500):
      data = b"X" * offset + b"HIT" + b"X" * (500 - offset)
      MockVFSHandlerFind.filesystem[self.filename] = data

      request = rdf_client_fs.GrepSpec(
          literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
          xor_in_key=self.XOR_IN_KEY,
          xor_out_key=self.XOR_OUT_KEY)
      request.target.path = self.filename
      request.target.pathtype = rdf_paths.PathSpec.PathType.OS
      request.start_offset = 0
      request.bytes_before = 10
      request.bytes_after = 10

      result = self.RunAction(searching.Grep, request)
      self.assertLen(result, 1)
      self.assertEqual(result[0].offset, offset)
      expected = data[max(0, offset - 10):offset + 3 + 10]
      self.assertLen(expected, result[0].length)
      self.assertEqual(utils.Xor(result[0].data, self.XOR_OUT_KEY), expected)

  def testHitLimit(self):
    limit = searching.Grep.HIT_LIMIT

    hit = b"x" * 10 + b"HIT" + b"x" * 10
    data = hit * (limit + 100)
    MockVFSHandlerFind.filesystem[self.filename] = data

    request = rdf_client_fs.GrepSpec(
        literal=utils.Xor(b"HIT", self.XOR_IN_KEY),
        xor_in_key=self.XOR_IN_KEY,
        xor_out_key=self.XOR_OUT_KEY)
    request.target.path = self.filename
    request.target.pathtype = rdf_paths.PathSpec.PathType.OS
    request.start_offset = 0
    request.bytes_before = 10
    request.bytes_after = 10

    result = self.RunAction(searching.Grep, request)
    self.assertLen(result, limit + 1)
    error = b"maximum number of hits"
    self.assertIn(error, utils.Xor(result[-1].data, self.XOR_OUT_KEY))


class XoredSearchingTest(GrepTest):
  """Test the searching client Actions using XOR."""

  XOR_IN_KEY = 37
  XOR_OUT_KEY = 57


def main(argv):
  test_lib.main(argv)


if __name__ == "__main__":
  app.run(main)
