Malte Ostendorff

import hashlib
import inspect
from datetime import datetime, timezone
from typing import Any, Optional, Tuple

from redb.extractors.enum import Tag
from redb.extractors.apk_extractor import APKExtractor
from redb.models.dataclasses import APKCertificate, APKCodeSigningInfo


# OID to RFC 4514 short name mapping
_OID_SHORT_NAMES = {
    '2.5.4.3': 'CN', '2.5.4.6': 'C', '2.5.4.7': 'L', '2.5.4.8': 'ST',
    '2.5.4.10': 'O', '2.5.4.11': 'OU', '1.2.840.113549.1.9.1': 'E',
    '2.5.4.5': 'SERIALNUMBER', '2.5.4.12': 'T', '2.5.4.42': 'GN',
    '2.5.4.4': 'SN', '2.5.4.9': 'STREET', '2.5.4.17': 'POSTALCODE',
}


def _asn1_name_to_rfc4514(name):
    """Convert an asn1crypto x509.Name to RFC 4514 string (CN=..., O=...)."""
    parts = []
    for rdn in name.chosen:
        for attr in rdn:
            oid = attr['type'].dotted
            val = attr['value'].native
            short = _OID_SHORT_NAMES.get(oid, oid)
            parts.append(f'{short}={val}')
    return ', '.join(parts)


class APKSignatureExtractor(APKExtractor):

    def __init__(
        self, filepath, log, exporters=None, index_prefix=None,
        known_benign=False, known_malicious=False,
        apk=None,
    ):
        super().__init__(
            filepath, log, exporters, index_prefix,
            known_benign, known_malicious, apk,
        )
        self.signing_info = None
        self.log.debug(inspect.currentframe().f_code.co_name)

    def tag(self):
        return Tag.APK_SIGNATURE.value

    def _detect_signature_schemes(self):
        """Detect which APK signature schemes are present."""
        schemes = []
        try:
            if self.apk.is_signed_v1():
                schemes.append(1)
        except Exception:
            pass
        try:
            if self.apk.is_signed_v2():
                schemes.append(2)
        except Exception:
            pass
        try:
            if self.apk.is_signed_v3():
                schemes.append(3)
        except Exception:
            pass
        return schemes

    def _extract_certificate_info(self, cert, source_scheme=None):
        """Extract certificate details from an androguard certificate object."""
        try:
            # Subject and Issuer — asn1crypto doesn't have rfc4514_string, convert manually
            subject = _asn1_name_to_rfc4514(cert.subject)
            issuer = _asn1_name_to_rfc4514(cert.issuer)

            # Serial number
            serial = str(cert.serial_number)

            # Validity
            valid_from = cert.not_valid_before
            valid_to = cert.not_valid_after
            if isinstance(valid_from, datetime):
                valid_from = valid_from.strftime("%Y-%m-%d %H:%M:%S")
            else:
                valid_from = str(valid_from)
            if isinstance(valid_to, datetime):
                valid_to = valid_to.strftime("%Y-%m-%d %H:%M:%S")
            else:
                valid_to = str(valid_to)

            # Thumbprints
            cert_der = cert.dump()
            thumbprint_sha1 = hashlib.sha1(cert_der).hexdigest()
            thumbprint_sha256 = hashlib.sha256(cert_der).hexdigest()

            # Algorithm
            algorithm = None
            try:
                algorithm = cert.hash_algo
            except Exception:
                try:
                    algorithm = cert.signature_algo
                except Exception:
                    pass

            # Key size
            key_size = None
            try:
                pub_key = cert.public_key
                if hasattr(pub_key, 'bit_size'):
                    key_size = pub_key.bit_size
                elif hasattr(pub_key, 'key_size'):
                    key_size = pub_key.key_size
            except Exception:
                pass

            # Self-signed check
            is_self_signed = cert.subject == cert.issuer

            return APKCertificate(
                subject=subject,
                issuer=issuer,
                serial_number=serial,
                valid_from=valid_from,
                valid_to=valid_to,
                thumbprint_sha1=thumbprint_sha1,
                thumbprint_sha256=thumbprint_sha256,
                algorithm=algorithm,
                key_size=key_size,
                is_self_signed=is_self_signed,
                certificate_source_scheme=source_scheme,
            )
        except Exception as e:
            self.log.error(f"Error extracting certificate info: {e}")
            return None

    def _get_signer_certificate(self, schemes) -> Tuple[Optional[APKCertificate], Optional[str]]:
        """Identify the actual signer certificate using V3>V2>V1 priority.

        V2/V3: certificates[0] is the signer per spec.
        V1: androguard's get_certificate() does SignerInfo matching internally.

        Returns (signer_cert_info, source_scheme) or (None, None).
        """
        # V3 takes priority (Android 9+)
        if 3 in schemes:
            try:
                certs = self.apk.get_certificates_v3()
                if certs:
                    cert_info = self._extract_certificate_info(certs[0], source_scheme='v3')
                    if cert_info:
                        return cert_info, 'v3'
            except Exception as e:
                self.log.debug(f"V3 signer extraction failed: {e}")

        # V2 next (Android 7.0+)
        if 2 in schemes:
            try:
                certs = self.apk.get_certificates_v2()
                if certs:
                    cert_info = self._extract_certificate_info(certs[0], source_scheme='v2')
                    if cert_info:
                        return cert_info, 'v2'
            except Exception as e:
                self.log.debug(f"V2 signer extraction failed: {e}")

        # V1 fallback — androguard does SignerInfo.issuerAndSerialNumber matching
        if 1 in schemes:
            try:
                names = self.apk.get_signature_names()
                if names:
                    cert = self.apk.get_certificate(names[0])
                    if cert:
                        cert_info = self._extract_certificate_info(cert, source_scheme='v1')
                        if cert_info:
                            return cert_info, 'v1'
            except Exception as e:
                self.log.debug(f"V1 signer extraction failed: {e}")

        return None, None

    def _extract_all_certificates(self, schemes):
        """Extract all certificates with their source scheme tracked.

        Uses scheme-specific methods to preserve source information,
        deduplicating by thumbprint_sha1 and merging source schemes.
        """
        # cert thumbprint -> (APKCertificate, set of source schemes)
        seen = {}

        for scheme_num, method_name in [
            (3, 'get_certificates_v3'),
            (2, 'get_certificates_v2'),
            (1, 'get_certificates_v1'),
        ]:
            if scheme_num not in schemes:
                continue
            try:
                method = getattr(self.apk, method_name, None)
                if method is None:
                    continue
                certs = method()
                if not certs:
                    continue
                for cert in certs:
                    cert_info = self._extract_certificate_info(cert, source_scheme=f'v{scheme_num}')
                    if cert_info:
                        key = cert_info.thumbprint_sha1
                        if key in seen:
                            # Merge source schemes
                            existing_cert, existing_schemes = seen[key]
                            existing_schemes.add(f'v{scheme_num}')
                        else:
                            seen[key] = (cert_info, {f'v{scheme_num}'})
            except Exception as e:
                self.log.warning(f"Error extracting v{scheme_num} certificates: {e}")

        # Build final list with merged source schemes
        result = []
        for cert_info, source_schemes in seen.values():
            cert_info.certificate_source_scheme = ','.join(sorted(source_schemes))
            result.append(cert_info)

        return result

    def extract(self):
        if not self._is_valid_apk():
            self.log.error(f"Invalid APK for {self.hash.sha256}")
            return None

        schemes = self._detect_signature_schemes()
        is_signed = len(schemes) > 0

        # Extract all certificates with source scheme tracking
        certificates = self._extract_all_certificates(schemes)

        # Identify the actual signer using V3>V2>V1 priority
        signer_cert, signer_scheme = self._get_signer_certificate(schemes)

        apk_signer_subject = signer_cert.subject if signer_cert else None
        apk_signer_issuer = signer_cert.issuer if signer_cert else None

        self.signing_info = APKCodeSigningInfo(
            _id=self.sha256,
            is_signed=is_signed,
            signature_scheme_versions=schemes,
            number_of_certificates=len(certificates),
            x509_certificates=certificates if certificates else None,
            apk_signer_subject=apk_signer_subject,
            apk_signer_issuer=apk_signer_issuer,
        )
        return self.signing_info

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ClickHouseExporter":
            if not self.signing_info:
                return None

            current_time = datetime.now(timezone.utc)
            sig = self.signing_info

            # Signature summary table
            signature_data = [[
                self.sha256,
                int(sig.is_signed),
                sig.signature_scheme_versions,
                sig.number_of_certificates,
                sig.apk_signer_subject,
                sig.apk_signer_issuer,
                current_time,
            ]]

            # Certificates table (one row per certificate)
            certificates_data = []
            if sig.x509_certificates:
                for cert in sig.x509_certificates:
                    certificates_data.append([
                        self.sha256,
                        cert.subject,
                        cert.issuer,
                        cert.serial_number,
                        datetime.strptime(cert.valid_from, "%Y-%m-%d %H:%M:%S") if cert.valid_from else None,
                        datetime.strptime(cert.valid_to, "%Y-%m-%d %H:%M:%S") if cert.valid_to else None,
                        cert.thumbprint_sha1,
                        cert.algorithm,
                        cert.key_size,
                        int(cert.is_self_signed),
                        cert.certificate_source_scheme,
                        current_time,
                    ])

            return {
                'multi_table': True,
                'signature': {
                    'table': 'redb_apk_signature',
                    'data': signature_data,
                    'column_names': [
                        'sha256',
                        'is_signed', 'signature_scheme_versions',
                        'number_of_certificates',
                        'apk_signer_subject', 'apk_signer_issuer',
                        'analysis_date',
                    ],
                    'column_type_names': [
                        'FixedString(64)',
                        'UInt8', 'Array(UInt8)',
                        'UInt8',
                        'Nullable(String)', 'Nullable(String)',
                        "DateTime64(3, 'UTC')",
                    ],
                },
                'certificates': {
                    'table': 'redb_apk_certificates',
                    'data': certificates_data,
                    'column_names': [
                        'sha256', 'certificate_subject', 'certificate_issuer',
                        'certificate_serial_number',
                        'certificate_valid_from', 'certificate_valid_to',
                        'certificate_thumbprint',
                        'certificate_algorithm', 'key_size', 'is_self_signed',
                        'certificate_source_scheme',
                        'analysis_date',
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'String', 'String', 'String',
                        "DateTime64(3, 'UTC')", "DateTime64(3, 'UTC')",
                        'FixedString(40)',
                        'LowCardinality(String)', 'Nullable(UInt16)', 'UInt8',
                        'LowCardinality(String)',
                        "DateTime64(3, 'UTC')",
                    ],
                },
            }

    def get_clickhouse_table(self) -> str:
        return "redb_apk_signature"