#!/usr/bin/env python
# -*- mode: python; encoding: utf-8 -*-
"""Unit test for the linux file parser."""

import operator
import os
import StringIO


from grr.lib import flags
from grr.lib import parsers
from grr.lib import utils
from grr.lib.rdfvalues import anomaly as rdf_anomaly
from grr.lib.rdfvalues import client as rdf_client
from grr.lib.rdfvalues import file_finder as rdf_file_finder
from grr.lib.rdfvalues import paths as rdf_paths
from grr.parsers import linux_file_parser
from grr.test_lib import test_lib


class LinuxFileParserTest(test_lib.GRRBaseTest):
  """Test parsing of linux files."""

  def testPCIDevicesInfoParser(self):
    """Ensure we can extract PCI devices info."""

    # Test when there's data for one PCI device only.
    test_data1 = {
        "/sys/bus/pci/devices/0000:00:01.0/vendor": "0x0e00\n",
        "/sys/bus/pci/devices/0000:00:01.0/class": "0x060400\n",
        "/sys/bus/pci/devices/0000:00:01.0/device": "0x0e02\n",
        "/sys/bus/pci/devices/0000:00:01.0/config": "0200"
    }
    device_1 = rdf_client.PCIDevice(
        domain=0,
        bus=0,
        device=1,
        function=0,
        class_id="0x060400",
        vendor="0x0e00",
        vendor_device_id="0x0e02",
        config="0200")
    parsed_results = self._ParsePCIDeviceTestData(test_data1)
    self._MatchPCIDeviceResultToExpected(parsed_results, [device_1])

    # Use raw bytes to test PCI device config works as expected.
    bytes2 = bytearray([234, 232, 231, 188, 122, 132, 145])
    test_data2 = {
        "/sys/bus/pci/devices/0000:00:00.0/vendor": "0x8086\n",
        "/sys/bus/pci/devices/0000:00:00.0/class": "0x060000\n",
        "/sys/bus/pci/devices/0000:00:00.0/device": "0x0e00\n",
        "/sys/bus/pci/devices/0000:00:00.0/config": bytes2
    }
    device_2 = rdf_client.PCIDevice(
        domain=0,
        bus=0,
        device=0,
        function=0,
        class_id="0x060000",
        vendor="0x8086",
        vendor_device_id="0x0e00",
        config=b"\xea\xe8\xe7\xbcz\x84\x91")
    parsed_results = self._ParsePCIDeviceTestData(test_data2)
    self._MatchPCIDeviceResultToExpected(parsed_results, [device_2])

    # Test for when there's missing data.
    test_data3 = {
        "/sys/bus/pci/devices/0000:00:03.0/vendor": "0x0e00\n",
        "/sys/bus/pci/devices/0000:00:03.0/config": "0030"
    }
    device_3 = rdf_client.PCIDevice(
        domain=0, bus=0, device=3, function=0, vendor="0x0e00", config="0030")
    parsed_results = self._ParsePCIDeviceTestData(test_data3)
    self._MatchPCIDeviceResultToExpected(parsed_results, [device_3])

    # Test when data contains non-valid B/D/F folders/files.
    test_data4 = {
        "/sys/bus/pci/devices/0000:00:05.0/vendor": "0x0e00\n",
        "/sys/bus/pci/devices/0000:00:05.0/class": "0x060400\n",
        "/sys/bus/pci/devices/0000:00:05.0/device": "0x0e02\n",
        "/sys/bus/pci/devices/0000:00:05.0/config": "0200",
        "/sys/bus/pci/devices/crazyrandomfile/test1": "test1",
        "/sys/bus/pci/devices/::./test2": "test2",
        "/sys/bus/pci/devices/00:5.0/test3": "test3"
    }
    device_4 = rdf_client.PCIDevice(
        domain=0,
        bus=0,
        device=5,
        function=0,
        class_id="0x060400",
        vendor="0x0e00",
        vendor_device_id="0x0e02",
        config="0200")
    parsed_results = self._ParsePCIDeviceTestData(test_data4)
    self._MatchPCIDeviceResultToExpected(parsed_results, [device_4])

    # Test when there's multiple PCI devices in the test_data.
    combined_data = test_data1.copy()
    combined_data.update(test_data3)
    combined_data.update(test_data4)
    combined_data.update(test_data2)
    parsed_results = self._ParsePCIDeviceTestData(combined_data)
    self._MatchPCIDeviceResultToExpected(
        parsed_results, [device_1, device_4, device_2, device_3])

  def _ParsePCIDeviceTestData(self, test_data):
    """Given test_data dictionary, parse it using PCIDevicesInfoParser."""
    parser = linux_file_parser.PCIDevicesInfoParser()
    stats = []
    file_objs = []
    kb_objs = []

    # Populate stats, file_ojbs, kb_ojbs lists needed by the parser.
    for filename, data in test_data.items():
      pathspec = rdf_paths.PathSpec(path=filename, pathtype="OS")
      stat = rdf_client.StatEntry(pathspec=pathspec)
      file_obj = StringIO.StringIO(data)
      stats.append(stat)
      file_objs.append(file_obj)
      kb_objs.append(None)

    return list(parser.ParseMultiple(stats, file_objs, kb_objs))

  def _MatchPCIDeviceResultToExpected(self, parsed_results, expected_output):
    """Make sure the parsed_results match expected_output."""

    # Check the size matches.
    self.assertEqual(len(parsed_results), len(expected_output))

    # Sort parsed_results and expected_outputs so we're comparing properly.
    results = sorted(parsed_results, key=operator.attrgetter("device"))
    outputs = sorted(expected_output, key=operator.attrgetter("device"))

    # Check all the content matches.
    for result, output in zip(results, outputs):
      self.assertEqual(result.domain, output.domain)
      self.assertEqual(result.bus, output.bus)
      self.assertEqual(result.device, output.device)
      self.assertEqual(result.function, output.function)
      self.assertEqual(result.class_id, output.class_id)
      self.assertEqual(result.vendor, output.vendor)
      self.assertEqual(result.vendor_device_id, output.vendor_device_id)
      self.assertEqual(result.config, output.config)

  def testPasswdParser(self):
    """Ensure we can extract users from a passwd file."""
    parser = linux_file_parser.PasswdParser()
    dat = """
user1:x:1000:1000:User1 Name,,,:/home/user1:/bin/bash
user2:x:1001:1001:User2 Name,,,:/home/user2:/bin/bash
"""
    out = list(parser.Parse(None, StringIO.StringIO(dat), None))
    self.assertEqual(len(out), 2)
    self.assertTrue(isinstance(out[1], rdf_client.User))
    self.assertTrue(isinstance(out[1], rdf_client.User))
    self.assertEqual(out[0].username, "user1")
    self.assertEqual(out[0].full_name, "User1 Name,,,")
    dat = """
user1:x:1000:1000:User1 Name,,,:/home/user1:/bin/bash
user2:x:1001:1001:User2 Name,,,:/home/user
"""
    parser = linux_file_parser.PasswdParser()
    self.assertRaises(parsers.ParseError, list,
                      parser.Parse(None, StringIO.StringIO(dat), None))

  def testPasswdBufferParser(self):
    """Ensure we can extract users from a passwd file."""
    parser = linux_file_parser.PasswdBufferParser()
    buf1 = rdf_client.BufferReference(data="user1:x:1000:1000:User1"
                                      " Name,,,:/home/user1:/bin/bash\n")

    buf2 = rdf_client.BufferReference(data="user2:x:1000:1000:User2"
                                      " Name,,,:/home/user2:/bin/bash\n")

    ff_result = rdf_file_finder.FileFinderResult(matches=[buf1, buf2])
    out = list(parser.Parse(ff_result, None))
    self.assertEqual(len(out), 2)
    self.assertTrue(isinstance(out[1], rdf_client.User))
    self.assertTrue(isinstance(out[1], rdf_client.User))
    self.assertEqual(out[0].username, "user1")
    self.assertEqual(out[0].full_name, "User1 Name,,,")

  def testNetgroupParser(self):
    """Ensure we can extract users from a netgroup file."""
    parser = linux_file_parser.NetgroupParser()
    dat = u"""group1 (-,user1,) (-,user2,) (-,user3,)
#group1 comment
group2 (-,user4,) (-,user2,)

super_group (-,user5,) (-,user6,) (-,文德文,) group1 group2
super_group2 (-,user7,) super_group
super_group3 (-,user5,) (-,user6,) group1 group2
"""
    dat_fd = StringIO.StringIO(dat)

    with test_lib.ConfigOverrider({
        "Artifacts.netgroup_user_blacklist": ["user2", "user3"]
    }):
      out = list(parser.Parse(None, dat_fd, None))
      users = []
      for result in out:
        if isinstance(result, rdf_anomaly.Anomaly):
          self.assertTrue(utils.SmartUnicode(u"文德文") in result.symptom)
        else:
          users.append(result)

      self.assertItemsEqual([x.username for x in users],
                            [u"user1", u"user4", u"user5", u"user6", u"user7"])

      dat_fd.seek(0)

    with test_lib.ConfigOverrider({
        "Artifacts.netgroup_filter_regexes": [r"^super_group3$"]
    }):
      out = list(parser.Parse(None, dat_fd, None))
      self.assertItemsEqual([x.username for x in out], [u"user5", u"user6"])

  def testNetgroupBufferParser(self):
    """Ensure we can extract users from a netgroup file."""
    parser = linux_file_parser.NetgroupBufferParser()
    buf1 = rdf_client.BufferReference(data="group1 (-,user1,) (-,user2,) "
                                      "(-,user3,)\n")
    buf2 = rdf_client.BufferReference(data="super_group3 (-,user5,) (-,user6,)"
                                      " group1 group2\n")

    ff_result = rdf_file_finder.FileFinderResult(matches=[buf1, buf2])
    with test_lib.ConfigOverrider({
        "Artifacts.netgroup_user_blacklist": ["user2", "user3"]
    }):
      out = list(parser.Parse(ff_result, None))
      self.assertItemsEqual([x.username
                             for x in out], [u"user1", u"user5", u"user6"])

  def testNetgroupParserBadInput(self):
    parser = linux_file_parser.NetgroupParser()
    dat = """group1 (-,user1,) (-,user2,) (-,user3,)
#group1 comment
group2 user4 (-user2,)
super_group (-,,user5,) (-user6,) group1 group2
super_group2 (-,user7,) super_group
"""
    self.assertRaises(parsers.ParseError, list,
                      parser.Parse(None, StringIO.StringIO(dat), None))

  def testWtmpParser(self):
    """Test parsing of wtmp file."""
    parser = linux_file_parser.LinuxWtmpParser()
    path = os.path.join(self.base_path, "VFSFixture/var/log/wtmp")
    with open(path, "rb") as wtmp_fd:
      out = list(parser.Parse(None, wtmp_fd, None))

    self.assertEqual(len(out), 3)
    self.assertItemsEqual(["%s:%d" % (x.username, x.last_logon) for x in out], [
        "user1:1296552099000000", "user2:1296552102000000",
        "user3:1296569997000000"
    ])


class LinuxShadowParserTest(test_lib.GRRBaseTest):
  """Test parsing of linux shadow files."""

  crypt = {
      "DES": "A.root/o0tr.o",
      "MD5": "$1$roo/root/o07r.0tROOTro",
      "SHA256": "$5$sal/s.lt5a17${0}".format("r" * 43),
      "SHA512": "$6$sa./sa1T${0}".format("r" * 86),
      "UNSET": "*",
      "DISABLED": "!$1$roo/rootroootROOO0oooTroooooooo",
      "EMPTY": ""
  }

  def _GenFiles(self, passwd, shadow, group, gshadow):
    stats = []
    files = []
    for path in ["/etc/passwd", "/etc/shadow", "/etc/group", "/etc/gshadow"]:
      p = rdf_paths.PathSpec(path=path)
      stats.append(rdf_client.StatEntry(pathspec=p))
    for data in passwd, shadow, group, gshadow:
      if data is None:
        data = []
      lines = "\n".join(data).format(**self.crypt)
      files.append(StringIO.StringIO(lines))
    return stats, files

  def testNoAnomaliesWhenEverythingIsFine(self):
    passwd = [
        "ok_1:x:1000:1000::/home/ok_1:/bin/bash",
        "ok_2:x:1001:1001::/home/ok_2:/bin/bash"
    ]
    shadow = [
        "ok_1:{SHA256}:16000:0:99999:7:::", "ok_2:{SHA512}:16000:0:99999:7:::"
    ]
    group = ["ok_1:x:1000:ok_1", "ok_2:x:1001:ok_2"]
    gshadow = ["ok_1:::ok_1", "ok_2:::ok_2"]
    stats, files = self._GenFiles(passwd, shadow, group, gshadow)
    parser = linux_file_parser.LinuxSystemPasswdParser()
    rdfs = parser.ParseMultiple(stats, files, None)
    results = [r for r in rdfs if isinstance(r, rdf_anomaly.Anomaly)]
    self.assertFalse(results)

  def testSystemGroupParserAnomaly(self):
    """Detect anomalies in group/gshadow files."""
    group = [
        "root:x:0:root,usr1", "adm:x:1:syslog,usr1",
        "users:x:1000:usr1,usr2,usr3,usr4"
    ]
    gshadow = ["root::usr4:root", "users:{DES}:usr1:usr2,usr3,usr4"]
    stats, files = self._GenFiles(None, None, group, gshadow)

    # Set up expected anomalies.
    member = {
        "symptom":
            "Group/gshadow members differ in group: root",
        "finding": [
            "Present in group, missing in gshadow: usr1",
            "Present in gshadow, missing in group: usr4"
        ],
        "type":
            "PARSER_ANOMALY"
    }
    group = {
        "symptom": "Mismatched group and gshadow files.",
        "finding": ["Present in group, missing in gshadow: adm"],
        "type": "PARSER_ANOMALY"
    }
    expected = [rdf_anomaly.Anomaly(**member), rdf_anomaly.Anomaly(**group)]

    parser = linux_file_parser.LinuxSystemGroupParser()
    rdfs = parser.ParseMultiple(stats, files, None)
    results = [r for r in rdfs if isinstance(r, rdf_anomaly.Anomaly)]

    self.assertEqual(len(expected), len(results))
    for expect, result in zip(expected, results):
      self.assertRDFValuesEqual(expect, result)

  def testSystemAccountAnomaly(self):
    passwd = [
        "root:x:0:0::/root:/bin/sash",
        "miss:x:1000:100:Missing:/home/miss:/bin/bash",
        "bad1:x:0:1001:Bad 1:/home/bad1:/bin/bash",
        "bad2:x:1002:0:Bad 2:/home/bad2:/bin/bash"
    ]
    shadow = [
        "root:{UNSET}:16000:0:99999:7:::", "ok:{SHA512}:16000:0:99999:7:::",
        "bad1::16333:0:99999:7:::", "bad2:{DES}:16333:0:99999:7:::"
    ]
    group = [
        "root:x:0:root", "miss:x:1000:miss", "bad1:x:1001:bad1",
        "bad2:x:1002:bad2"
    ]
    gshadow = ["root:::root", "miss:::miss", "bad1:::bad1", "bad2:::bad2"]
    stats, files = self._GenFiles(passwd, shadow, group, gshadow)

    no_grp = {
        "symptom": "Accounts with invalid gid.",
        "finding": ["gid 100 assigned without /etc/groups entry: miss"],
        "type": "PARSER_ANOMALY"
    }
    uid = {
        "symptom": "Accounts with shared uid.",
        "finding": ["uid 0 assigned to multiple accounts: bad1,root"],
        "type": "PARSER_ANOMALY"
    }
    gid = {
        "symptom": "Privileged group with unusual members.",
        "finding": ["Accounts in 'root' group: bad2"],
        "type": "PARSER_ANOMALY"
    }
    no_match = {
        "symptom":
            "Mismatched passwd and shadow files.",
        "finding": [
            "Present in passwd, missing in shadow: miss",
            "Present in shadow, missing in passwd: ok"
        ],
        "type":
            "PARSER_ANOMALY"
    }
    expected = [
        rdf_anomaly.Anomaly(**no_grp),
        rdf_anomaly.Anomaly(**uid),
        rdf_anomaly.Anomaly(**gid),
        rdf_anomaly.Anomaly(**no_match)
    ]

    parser = linux_file_parser.LinuxSystemPasswdParser()
    rdfs = parser.ParseMultiple(stats, files, None)
    results = [r for r in rdfs if isinstance(r, rdf_anomaly.Anomaly)]

    self.assertEqual(len(expected), len(results))
    for expect, result in zip(expected, results):
      self.assertEqual(expect.symptom, result.symptom)
      # Expand out repeated field helper.
      self.assertItemsEqual(list(expect.finding), list(result.finding))
      self.assertEqual(expect.type, result.type)

  def GetExpectedUser(self, algo, user_store, group_store):
    user = rdf_client.User(
        username="user",
        full_name="User",
        uid="1001",
        gid="1001",
        homedir="/home/user",
        shell="/bin/bash")
    user.pw_entry = rdf_client.PwEntry(store=user_store, hash_type=algo)
    user.gids = [1001]
    grp = rdf_client.Group(gid=1001, members=["user"], name="user")
    grp.pw_entry = rdf_client.PwEntry(store=group_store, hash_type=algo)
    return user, grp

  def CheckExpectedUser(self, algo, expect, result):
    self.assertEqual(expect.username, result.username)
    self.assertEqual(expect.gid, result.gid)
    self.assertEqual(expect.pw_entry.store, result.pw_entry.store)
    self.assertEqual(expect.pw_entry.hash_type, result.pw_entry.hash_type)
    self.assertItemsEqual(expect.gids, result.gids)

  def CheckExpectedGroup(self, algo, expect, result):
    self.assertEqual(expect.name, result.name)
    self.assertEqual(expect.gid, result.gid)
    self.assertEqual(expect.pw_entry.store, result.pw_entry.store)
    self.assertEqual(expect.pw_entry.hash_type, result.pw_entry.hash_type)

  def CheckCryptResults(self, passwd, shadow, group, gshadow, algo, usr, grp):
    stats, files = self._GenFiles(passwd, shadow, group, gshadow)
    parser = linux_file_parser.LinuxSystemPasswdParser()
    results = list(parser.ParseMultiple(stats, files, None))
    usrs = [r for r in results if isinstance(r, rdf_client.User)]
    grps = [r for r in results if isinstance(r, rdf_client.Group)]
    self.assertEqual(1, len(usrs), "Different number of usr %s results" % algo)
    self.assertEqual(1, len(grps), "Different number of grp %s results" % algo)
    self.CheckExpectedUser(algo, usr, usrs[0])
    self.CheckExpectedGroup(algo, grp, grps[0])

  def testSetShadowedEntries(self):
    passwd = ["user:x:1001:1001:User:/home/user:/bin/bash"]
    group = ["user:x:1001:user"]
    for algo, crypted in self.crypt.iteritems():
      # Flush the parser for each iteration.
      shadow = ["user:%s:16000:0:99999:7:::" % crypted]
      gshadow = ["user:%s::user" % crypted]
      usr, grp = self.GetExpectedUser(algo, "SHADOW", "GSHADOW")
      self.CheckCryptResults(passwd, shadow, group, gshadow, algo, usr, grp)

  def testSetNonShadowedEntries(self):
    shadow = ["user::16000:0:99999:7:::"]
    gshadow = ["user:::user"]
    for algo, crypted in self.crypt.iteritems():
      # Flush the parser for each iteration.
      passwd = ["user:%s:1001:1001:User:/home/user:/bin/bash" % crypted]
      group = ["user:%s:1001:user" % crypted]
      usr, grp = self.GetExpectedUser(algo, "PASSWD", "GROUP")
      self.CheckCryptResults(passwd, shadow, group, gshadow, algo, usr, grp)


class LinuxDotFileParserTest(test_lib.GRRBaseTest):
  """Test parsing of user dotfiles."""

  def testFindPaths(self):
    # TODO(user): Deal with cases where multiple vars are exported.
    # export TERM PERLLIB=.:shouldntbeignored
    bashrc_data = StringIO.StringIO("""
      IGNORE='bad' PATH=${HOME}/bin:$PATH
     { PYTHONPATH=/path1:/path2 }
      export TERM=screen-256color
      export http_proxy="http://proxy.example.org:3128/"
      export HTTP_PROXY=$http_proxy
      if [[ "$some_condition" ]]; then
        export PATH=:$PATH; LD_LIBRARY_PATH=foo:bar:$LD_LIBRARY_PATH
        PYTHONPATH=$PATH:"${PYTHONPATH}"
        CLASSPATH=
      fi
      echo PATH=/should/be/ignored
      # Ignore PATH=foo:bar
      TERM=vt100 PS=" Foo" PERL5LIB=:shouldntbeignored
    """)
    cshrc_data = StringIO.StringIO("""
      setenv PATH ${HOME}/bin:$PATH
      setenv PYTHONPATH /path1:/path2
      set term = (screen-256color)
      setenv http_proxy "http://proxy.example.org:3128/"
      setenv HTTP_PROXY $http_proxy
      if ( -e "$some_condition" ) then
        set path =  (. $path); setenv LD_LIBRARY_PATH foo:bar:$LD_LIBRARY_PATH
        setenv PYTHONPATH $PATH:"${PYTHONPATH}"
        setenv CLASSPATH
      endif
      echo PATH=/should/be/ignored
      setenv PERL5LIB :shouldntbeignored
    """)
    parser = linux_file_parser.PathParser()
    bashrc_stat = rdf_client.StatEntry(pathspec=rdf_paths.PathSpec(
        path="/home/user1/.bashrc", pathtype="OS"))
    cshrc_stat = rdf_client.StatEntry(pathspec=rdf_paths.PathSpec(
        path="/home/user1/.cshrc", pathtype="OS"))
    bashrc = {
        r.name: r.vals
        for r in parser.Parse(bashrc_stat, bashrc_data, None)
    }
    cshrc = {r.name: r.vals for r in parser.Parse(cshrc_stat, cshrc_data, None)}
    expected = {
        "PATH": [".", "${HOME}/bin", "$PATH"],
        "PYTHONPATH": [".", "${HOME}/bin", "$PATH", "/path1", "/path2"],
        "LD_LIBRARY_PATH": ["foo", "bar", "$LD_LIBRARY_PATH"],
        "CLASSPATH": [],
        "PERL5LIB": [".", "shouldntbeignored"]
    }
    # Got the same environment variables for bash and cshrc files.
    self.assertItemsEqual(expected, bashrc)
    self.assertItemsEqual(expected, cshrc)
    # The path values are expanded correctly.
    for var_name in ("PATH", "PYTHONPATH", "LD_LIBRARY_PATH"):
      self.assertEqual(expected[var_name], bashrc[var_name])
      self.assertEqual(expected[var_name], cshrc[var_name])


def main(args):
  test_lib.main(args)


if __name__ == "__main__":
  flags.StartMain(main)
