#!/usr/bin/env python
#
# Copyright (C) 2019 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""deapexer is a tool that prints out content of an APEX.

To print content of an APEX to stdout:
  deapexer list foo.apex

To extract content of an APEX to the given directory:
  deapexer extract foo.apex dest
"""
from __future__ import print_function

import argparse
import apex_manifest
import enum
import os
import re
import shutil
import sys
import subprocess
import tempfile
import zipfile

BLOCK_SIZE = 4096

# See apexd/apex_file.cpp#RetrieveFsType
FS_TYPES = [
    ('f2fs', 1024, b'\x10\x20\xf5\xf2'),
    ('ext4', 1024 + 0x38, b'\123\357'),
    ('erofs', 1024, b'\xe2\xe1\xf5\xe0'),
]


def RetrieveFileSystemType(file):
  """Returns filesystem type with magic"""
  with open(file, 'rb') as f:
    for fs_type, offset, magic in FS_TYPES:
      buf = bytearray(len(magic))
      f.seek(offset, os.SEEK_SET)
      f.readinto(buf)
      if buf == magic:
        return fs_type
  raise ValueError('Failed to retrieve filesystem type')

class ApexImageEntry(object):
  """Represents an entry in APEX payload"""
  def __init__(self, name, *, base_dir, permissions, size, ino, extents,
               is_directory, is_symlink, security_context):
    self._name = name
    self._base_dir = base_dir
    self._permissions = permissions
    self._size = size
    self._is_directory = is_directory
    self._is_symlink = is_symlink
    self._ino = ino
    self._extents = extents
    self._security_context = security_context
    self._entries = []

  @property
  def name(self):
    return self._name

  @property
  def root(self):
    return self._base_dir == './' and self._name == '.'

  @property
  def full_path(self):
    if self.root:
      return self._base_dir  # './'
    path = os.path.join(self._base_dir, self._name)
    if self.is_directory:
      path += '/'
    return path

  @property
  def is_directory(self):
    return self._is_directory

  @property
  def is_symlink(self):
    return self._is_symlink

  @property
  def is_regular_file(self):
    return not self.is_directory and not self.is_symlink

  @property
  def permissions(self):
    return self._permissions

  @property
  def size(self):
    return self._size

  @property
  def ino(self):
    return self._ino

  @property
  def entries(self):
    return self._entries

  @property
  def extents(self):
    return self._extents

  @property
  def security_context(self):
    return self._security_context

  def __str__(self):
    ret = ''
    if self._is_directory:
      ret += 'd'
    elif self._is_symlink:
      ret += 'l'
    else:
      ret += '-'

    def MaskAsString(m):
      ret = 'r' if m & 4 == 4 else '-'
      ret += 'w' if m & 2 == 2 else '-'
      ret += 'x' if m & 1 == 1 else '-'
      return ret

    ret += MaskAsString(self._permissions >> 6)
    ret += MaskAsString((self._permissions >> 3) & 7)
    ret += MaskAsString(self._permissions & 7)

    return ret + ' ' + self._size + ' ' + self._name


class Apex(object):
  """Represents an APEX file"""
  def __init__(self, args):
    self._debugfs = args.debugfs_path
    self._fsckerofs = args.fsckerofs_path
    self._apex = args.apex
    self._tempdir = tempfile.mkdtemp()
    with zipfile.ZipFile(self._apex, 'r') as zip_ref:
      self._payload = zip_ref.extract('apex_payload.img', path=self._tempdir)
    self._payload_fs_type = RetrieveFileSystemType(self._payload)

  def __del__(self):
    shutil.rmtree(self._tempdir)

  def __enter__(self):
    return self

  def __exit__(self, ex_type, value, traceback):
    pass

  def list(self):
    if self._payload_fs_type not in ['ext4']:
      sys.exit(f'{self._payload_fs_type} is not supported for `list`.')

    yield from self.entries()

  def read_dir(self, path) -> ApexImageEntry:
    assert path.endswith('/')
    assert self.payload_fs_type == 'ext4'

    res = subprocess.check_output([self._debugfs, '-R', f'ls -l -p {path}', self._payload],
                                  text=True, stderr=subprocess.DEVNULL)
    dir_entry = None
    entries = []
    for line in res.split('\n'):
      if not line:
        continue
      parts = line.split('/')
      if len(parts) != 8:
        continue
      name = parts[5]
      if not name:
        continue
      if name == '..':
        continue
      if name == 'lost+found' and path == './':
        continue
      ino = parts[1]
      bits = parts[2]
      size = parts[6]
      extents = []
      is_symlink = bits[1]=='2'
      is_directory=bits[1]=='4'

      if not is_symlink and not is_directory:
        stdout = subprocess.check_output([self._debugfs, '-R', f'dump_extents <{ino}>',
                                          self._payload], text=True, stderr=subprocess.DEVNULL)
        # Output of dump_extents for an inode fragmented in 3 blocks (length and addresses represent
        # block-sized sections):
        # Level Entries       Logical      Physical Length Flags
        # 0/ 0   1/  3     0 -     0    18 -    18      1
        # 0/ 0   2/  3     1 -    15    20 -    34     15
        # 0/ 0   3/  3    16 -  1863    37 -  1884   1848
        res = stdout.splitlines()
        res.pop(0) # the first line contains only columns names
        left_length = int(size)
        try: # dump_extents sometimes has an unexpected output
          for line in res:
            tokens = line.split()
            offset = int(tokens[7]) * BLOCK_SIZE
            length = min(int(tokens[-1]) * BLOCK_SIZE, left_length)
            left_length -= length
            extents.append((offset, length))
          if left_length != 0: # dump_extents sometimes fails to display "hole" blocks
            raise ValueError
        except: # pylint: disable=bare-except
          extents = [] # [] means that we failed to retrieve the file location successfully

      # get 'security.selinux' attribute
      entry_path = os.path.join(path, name)
      stdout = subprocess.check_output([
        self._debugfs,
        '-R',
        f'ea_get -V {entry_path} security.selinux',
        self._payload
      ], text=True, stderr=subprocess.DEVNULL)
      security_context = stdout.rstrip('\n\x00')

      entry = ApexImageEntry(name,
                             base_dir=path,
                             permissions=int(bits[3:], 8),
                             size=size,
                             is_directory=is_directory,
                             is_symlink=is_symlink,
                             ino=ino,
                             extents=extents,
                             security_context=security_context)
      if name == '.':
        dir_entry = entry
      elif is_directory:
        sub_dir_entry = self.read_dir(path + name + '/')
        # sub_dir_entry should be the same inode
        assert entry.ino == sub_dir_entry.ino
        entry.entries.extend(sub_dir_entry.entries)
        entries.append(entry)
      else:
        entries.append(entry)

    assert dir_entry
    dir_entry.entries.extend(sorted(entries, key=lambda e: e.name))
    return dir_entry

  def extract(self, dest):
    """Recursively dumps contents of the payload with retaining mode bits, but not owner/group"""
    if self._payload_fs_type == 'erofs':
      subprocess.run([self._fsckerofs, f'--extract={dest}', '--overwrite',
                     '--no-preserve-owner', self._payload], stdout=subprocess.DEVNULL, check=True)
    elif self._payload_fs_type == 'ext4':
      # Extract entries one by one using `dump` because `rdump` doesn't support
      # "no-perserve" mode
      for entry in self.entries():
        self.write_entry(entry, dest)
    else:
      # TODO(b/279688635) f2fs is not supported yet.
      sys.exit(f'{self._payload_fs_type} is not supported for `extract`.')

  @property
  def payload_fs_type(self) -> str:
    return self._payload_fs_type

  def entries(self):
    """Generator to visit all entries in the payload starting from root(./)"""

    def TopDown(entry):
      yield entry
      for child in entry.entries:
        yield from TopDown(child)

    root = self.read_dir('./')
    yield from TopDown(root)

  def read_symlink(self, entry):
    assert entry.is_symlink
    assert self.payload_fs_type == 'ext4'

    stdout = subprocess.check_output([self._debugfs, '-R', f'stat {entry.full_path}',
                                      self._payload], text=True, stderr=subprocess.DEVNULL)
    # Output of stat for a symlink should have the following line:
    #   Fast link dest: \"%.*s\"
    m = re.search(r'\bFast link dest: \"(.+)\"\n', stdout)
    if m:
      return m.group(1)

    # if above match fails, it means it's a slow link. Use cat.
    output = subprocess.check_output([self._debugfs, '-R', f'cat {entry.full_path}',
                                      self._payload], text=True, stderr=subprocess.DEVNULL)

    if not output:
      sys.exit('failed to read symlink target')
    return output

  def write_entry(self, entry, out_dir):
    dest = os.path.normpath(os.path.join(out_dir, entry.full_path))
    if entry.is_directory:
      if not os.path.exists(dest):
        os.makedirs(dest, mode=0o755)
    elif entry.is_symlink:
      os.symlink(self.read_symlink(entry), dest)
    else:
      subprocess.check_output([self._debugfs, '-R', f'dump {entry.full_path} {dest}',
        self._payload], text=True, stderr=subprocess.DEVNULL)
      # retain mode bits
      os.chmod(dest, entry.permissions)


def RunList(args):
  if GetType(args.apex) == ApexType.COMPRESSED:
    with tempfile.TemporaryDirectory() as temp:
      decompressed_apex = os.path.join(temp, 'temp.apex')
      Decompress(args.apex, decompressed_apex)
      args.apex = decompressed_apex

      RunList(args)
      return

  with Apex(args) as apex:
    for e in apex.list():
      # dot(., ..) directories
      if not e.root and e.name in ('.', '..'):
        continue
      res = ''
      if args.size:
        res += e.size + ' '
      res += e.full_path
      if args.extents:
        res += ' [' + '-'.join(str(x) for x in e.extents) + ']'
      if args.contexts:
        res += ' ' + e.security_context
      print(res)


def RunExtract(args):
  if GetType(args.apex) == ApexType.COMPRESSED:
    with tempfile.TemporaryDirectory() as temp:
      decompressed_apex = os.path.join(temp, 'temp.apex')
      Decompress(args.apex, decompressed_apex)
      args.apex = decompressed_apex

      RunExtract(args)
      return

  with Apex(args) as apex:
    if not os.path.exists(args.dest):
      os.makedirs(args.dest, mode=0o755)
    apex.extract(args.dest)
    if os.path.isdir(os.path.join(args.dest, 'lost+found')):
      shutil.rmtree(os.path.join(args.dest, 'lost+found'))

class ApexType(enum.Enum):
  INVALID = 0
  UNCOMPRESSED = 1
  COMPRESSED = 2


def GetType(apex_path):
  with zipfile.ZipFile(apex_path, 'r') as zip_file:
    names = zip_file.namelist()
    has_payload = 'apex_payload.img' in names
    has_original_apex = 'original_apex' in names
    if has_payload and has_original_apex:
      return ApexType.INVALID
    if has_payload:
      return ApexType.UNCOMPRESSED
    if has_original_apex:
      return ApexType.COMPRESSED
    return ApexType.INVALID


def RunInfo(args):
  if args.print_type:
    res = GetType(args.apex)
    if res == ApexType.INVALID:
      print(args.apex + ' is not a valid apex')
      sys.exit(1)
    print(res.name)
  elif args.print_payload_type:
    print(Apex(args).payload_fs_type)
  else:
    manifest = apex_manifest.fromApex(args.apex)
    print(apex_manifest.toJsonString(manifest))


def RunDecompress(args):
  """RunDecompress takes path to compressed APEX and decompresses it to
  produce the original uncompressed APEX at give output path

  See apex_compression_tool.py#RunCompress for details on compressed APEX
  structure.

  Args:
      args.input: file path to compressed APEX
      args.output: file path to where decompressed APEX will be placed
  """
  if GetType(args.input) == ApexType.UNCOMPRESSED and args.copy_if_uncompressed:
    shutil.copyfile(args.input, args.output)
    return

  compressed_apex_fp = args.input
  decompressed_apex_fp = args.output
  return Decompress(compressed_apex_fp, decompressed_apex_fp)


def Decompress(compressed_apex_fp, decompressed_apex_fp):
  if os.path.exists(decompressed_apex_fp):
    print("Output path '" + decompressed_apex_fp + "' already exists")
    sys.exit(1)

  with zipfile.ZipFile(compressed_apex_fp, 'r') as zip_obj:
    if 'original_apex' not in zip_obj.namelist():
      print(compressed_apex_fp + ' is not a compressed APEX. Missing '
                                 "'original_apex' file inside it.")
      sys.exit(1)
    # Rename original_apex file to what user provided as output filename
    original_apex_info = zip_obj.getinfo('original_apex')
    original_apex_info.filename = os.path.basename(decompressed_apex_fp)
    # Extract the original_apex as desired name
    zip_obj.extract(original_apex_info,
                    path=os.path.dirname(decompressed_apex_fp))


def main(argv):
  parser = argparse.ArgumentParser()

  debugfs_default = None
  fsckerofs_default = None
  if 'ANDROID_HOST_OUT' in os.environ:
    debugfs_default = os.path.join(os.environ['ANDROID_HOST_OUT'], 'bin/debugfs_static')
    fsckerofs_default = os.path.join(os.environ['ANDROID_HOST_OUT'], 'bin/fsck.erofs')
  parser.add_argument(
      '--debugfs_path', help='The path to debugfs binary', default=debugfs_default)
  parser.add_argument(
      '--fsckerofs_path', help='The path to fsck.erofs binary', default=fsckerofs_default)
  # TODO(b/279858383) remove the argument
  parser.add_argument('--blkid_path', help='NOT USED')

  subparsers = parser.add_subparsers(required=True, dest='cmd')

  parser_list = subparsers.add_parser(
      'list', help='prints content of an APEX to stdout')
  parser_list.add_argument('apex', type=str, help='APEX file')
  parser_list.add_argument(
      '--size', help='also show the size of the files', action='store_true')
  parser_list.add_argument(
      '--extents', help='also show the location of the files', action='store_true')
  parser_list.add_argument('-Z', '--contexts',
                           help='also show the security context of the files',
                           action='store_true')
  parser_list.set_defaults(func=RunList)

  parser_extract = subparsers.add_parser('extract', help='extracts content of an APEX to the given '
                                                         'directory')
  parser_extract.add_argument('apex', type=str, help='APEX file')
  parser_extract.add_argument('dest', type=str, help='Directory to extract content of APEX to')
  parser_extract.set_defaults(func=RunExtract)

  parser_info = subparsers.add_parser('info', help='prints APEX manifest')
  parser_info.add_argument('apex', type=str, help='APEX file')
  parser_info.add_argument('--print-type',
                           help='Prints type of the apex (COMPRESSED or UNCOMPRESSED)',
                           action='store_true')
  parser_info.add_argument('--print-payload-type',
                           help='Prints filesystem type of the apex payload',
                           action='store_true')
  parser_info.set_defaults(func=RunInfo)

  # Handle sub-command "decompress"
  parser_decompress = subparsers.add_parser('decompress',
                                            help='decompresses a compressed '
                                                 'APEX')
  parser_decompress.add_argument('--input', type=str, required=True,
                                 help='path to compressed APEX file that '
                                      'will be decompressed')
  parser_decompress.add_argument('--output', type=str, required=True,
                                 help='path to the output APEX file')
  parser_decompress.add_argument('--copy-if-uncompressed',
                                 help='just copy the input if not compressed',
                                 action='store_true')
  parser_decompress.set_defaults(func=RunDecompress)

  args = parser.parse_args(argv)

  debugfs_required_for_cmd = ['list', 'extract']
  if args.cmd in debugfs_required_for_cmd and not args.debugfs_path:
    print('ANDROID_HOST_OUT environment variable is not defined, --debugfs_path must be set',
          file=sys.stderr)
    sys.exit(1)

  if args.cmd == 'extract':
    if not args.fsckerofs_path:
      print('ANDROID_HOST_OUT environment variable is not defined, --fsckerofs_path must be set',
            file=sys.stderr)
      sys.exit(1)

    if not os.path.isfile(args.fsckerofs_path):
      print(f'Cannot find fsck.erofs specified at {args.fsckerofs_path}',
            file=sys.stderr)
      sys.exit(1)

  args.func(args)


if __name__ == '__main__':
  main(sys.argv[1:])