Мы подготовили перевод статьи Райана Сирса об обработке логов Google’s Certificate Transparency, состоящей из двух частей. В первой части дается общее представление о структуре логов и приводится пример кода на Python для парсинга записей из этих логов. Вторая часть посвящена получению всех сертификатов из доступных логов и настройке системы Google BigQuery для хранения и организации поиска по полученным данным.

С момента написания оригинала прошло уже три года и с тех пор количество доступных логов и, соответственно, записей в них возросло многократно. Тем более важно правильно подойти к обработке логов, если целью является максимизация количества получаемых данных.

Часть 1. Parsing Certificate Transparency Logs Like a Boss

В процессе разработки нашего первого проекта - phisfinder, я потратил много времени размышляя об анатомии фишинговых атак и источниках данных, которые позволяли бы идентифицировать следы, указывающие на предстоящие фишинговые кампаний еще до того, как они смогут нанести какой-либо реальный ущерб.

Одним из источников, которые мы интегрировали (и определённо одним из лучших), стал Certificate Transparency Log (CTL) - проект, начатый Беном Лори и Адамом Ленгли в Google. По сути CTL - это лог, содержащий неизменяемый список сертификатов, выпущенных CA, который хранится в дереве Меркла, что позволяет при необходимости криптографически проверить каждый сертификат.

Чтобы понять, с какими объемами данных придется иметь дело, давайте посмотрим, сколько записей содержится в каждом логе из списка с сайта CTL:

import requests
import json
import locale
locale.setlocale(locale.LC_ALL, 'en_US')

ctl_log = requests.get(
    'https://www.gstatic.com/ct/log_list/log_list.json'
).json()

total_certs = 0

human_format = lambda x: locale.format('%d', x, grouping=True)

for log in ctl_log['logs']:
    log_url = log['url']
    try:
        log_info = requests.get(
            'https://{}/ct/v1/get-sth'.format(log_url),
            timeout=3
        ).json()
        total_certs += int(log_info['tree_size'])
    except:
        continue

    print("{} has {} certificates".format(
        log_url,
        human_format(log_info['tree_size'])
    ))

print("Total certs -> {}".format(human_format(total_certs)))

На выходе получаем:

ct.googleapis.com/pilot has 92,224,404 certificates
ct.googleapis.com/aviator has 46,466,472 certificates
ct1.digicert-ct.com/log has 1,577,183 certificates
ct.googleapis.com/rocketeer has 89,391,361 certificates
ct.ws.symantec.com has 3,562,198 certificates
ctlog.api.venafi.com has 94,797 certificates
vega.ws.symantec.com has 200,401 certificates
ctserver.cnnic.cn has 5,081 certificates
ctlog.wosign.com has 1,387,492 certificates
ct.startssl.com has 293,374 certificates
ct.googleapis.com/skydiver has 1,249,079 certificates
ct.googleapis.com/icarus has 48,585,765 certificates
Total certs -> 285,037,607

285,037,607 на момент написания статьи. Это не такой большой объем данных, но все равно придется приложить определенные усилия, чтобы эффективно организовать хранение и поиск по сертификатам. Подробнее об этом во второй части.

Анатомия CTL

Получение записей из CTL осуществляется по HTTP, что позволит нам легко получать данные с помощью современных библиотек. К сожалению, сами данные в записях представляют собой запутанные бинарные структуры, что несколько усложняет процесс парсинга. Пример записи в логе:

// curl -s 'https://ct1.digicert-ct.com/log/ct/v1/get-entries?start=0&end=0' | jq .
{
  "entries": [
    {
      "leaf_input": "AAAAAAFIyfaldAAAAAcDMIIG/zCCBeegAwIBAgI...",
      "extra_data": "AAiJAAS6MIIEtjCCA56gAwIBAgIQDHmpRLCMEZU..."
    }
  ]
}

Каждая запись содержит поля leaf_input и extra_data в формате base64. Обращаясь к RFC6962 видим, что leaf_input - закодированная структура MerkleTreeLeaf, а extra_data - PrecertChainEntry.

Про PreCerts

Мне потребовалось довольно много времени, чтобы разобраться, что вообще такое PreCert (можете попробовать сами, почитайте RFC, и, по всей видимости, я такой не один. Сохраню вам кучу времени на раздумья и поиски в гугле и сформулирую назначение PreCerts следующим образом:

PreCerts это отдельный тип сертификата, выпускаемого CA до того, как тот выпустит “настоящий” сертификат. Фактически, это копия исходного сертификата, но содержащая специальное расширение x509 v3, называемое poison и отмеченное как критическое. Таким образом, сертификат не будет валидирован как платформами, распознающими это расширение, и знающими что это PreCert, так и платформами, которые это расширение не распознают.

Мой опыт в ИБ говорит о том, что такая мера не сильно эффективна, хотя бы потому, что баги при парсинге x509/ASN.1 встречаются довольно часто и отдельные реализации могут быть уязвимы к различным махинациям, которые в конечном счете позволят валидировать PreCert. Я понимаю, зачем это было сделано, но складывается ощущение, что полностью убрать PreCerts и оставлять в CTL только сертификаты, действительно выпущенные CA, было бы намного разумнее.

Парсим бинарные структуры

Как для человека, занимающегося реверс-инжинирингом и время от времени участвующего в разных CTF, задача парсинга бинарных структур для меня не нова. Большинство людей в таких случаях обращаются к модулю struct, но много лет назад, во время работы на Филлипа Мартина, он познакомил меня с отличной библиотекой Construct, которая заметно упрощает парсинг подобных структур. Ниже приведены структуры, которые я использовал для парсинга, а также пример их использования для обработки записей:

from construct import     Struct, Byte, Int16ub, Int64ub, Enum, Bytes, Int24ub, this,     GreedyBytes, GreedyRange, Terminated, Embedded

MerkleTreeHeader = Struct(
    "Version"         / Byte,
    "MerkleLeafType"  / Byte,
    "Timestamp"       / Int64ub,
    "LogEntryType"    / Enum(Int16ub, X509LogEntryType=0, PrecertLogEntryType=1),
    "Entry"           / GreedyBytes
)

Certificate = Struct(
    "Length" / Int24ub,
    "CertData" / Bytes(this.Length)
)

CertificateChain = Struct(
    "ChainLength" / Int24ub,
    "Chain" / GreedyRange(Certificate),
)

PreCertEntry = Struct(
    "LeafCert" / Certificate,
    Embedded(CertificateChain),
    Terminated
)

import json
import base64

import ctl_parser_structures

from OpenSSL import crypto

entry = json.loads("""
{
  "entries": [
    {
      "leaf_input": "AAAAAAFIyfaldAAAAAcDMIIG/zCCBeegAwIBAgIQ...",
      "extra_data": "AAiJAAS6MIIEtjCCA56gAwIBAgIQDHmpRLCMEZUg..."
    }
  ]
}
""")['entries'][0]

leaf_cert = ctl_parser_structures.MerkleTreeHeader.parse(
    base64.b64decode(entry['leaf_input'])
)

print("Leaf Timestamp: {}".format(leaf_cert.Timestamp))
print("Entry Type: {}".format(leaf_cert.LogEntryType))

if leaf_cert.LogEntryType == "X509LogEntryType":
    # В случае, если запись - обычный X509 сертификат
    cert_data_string = ctl_parser_structures.Certificate.parse(
        leaf_cert.Entry).CertData
    chain = [
        crypto.load_certificate(crypto.FILETYPE_ASN1, cert_data_string)
    ]

    # Парсим структуру `extra_data`
    # чтобы получить оставшуюся часть цепочки
    extra_data = ctl_parser_structures.CertificateChain.parse(
        base64.b64decode(entry['extra_data'])
    )
    for cert in extra_data.Chain:
        chain.append(
            crypto.load_certificate(crypto.FILETYPE_ASN1, cert.CertData)
        )
else:
    #  В случае, если запись - PreCert
    extra_data = ctl_parser_structures.PreCertEntry.parse(
        base64.b64decode(entry['extra_data'])
    )
    chain = [
        crypto.load_certificate(
            crypto.FILETYPE_ASN1, extra_data.LeafCert.CertData
        )
    ]

    for cert in extra_data.Chain:
        chain.append(
            crypto.load_certificate(crypto.FILETYPE_ASN1, cert.CertData)
        )

Получаем массив X509 сертификатов из цепочки с сертификатом из leaf_input в качестве первого элемента

Как можете заметить, Construct позволяет довольно легко определять бинарные структуры на Python.

Теперь, понимая, что такое CTL и как парсить отдельные записи, можем переходить ко второй части - получению и сохранению всех записей из логов с возможностью последующего поиска по сертификатам.

Часть 2. Retrieving, Storing and Querying 250M+ Certificates Like a Boss

Сбор сертификатов

В соответствии с RFC, для получения записей из логов используется эндпоинт get-entries. К сожалению, задача осложняется ограничением на максимальное количество записей, которое можно получить в одном запросе (контролируется параметрами start и end), и большинство логов позволяют получить лишь 64 записи за раз. Однако CTL от Google, составляющие большинство всех логов, используют максимальный размер запроса в 1024 записи.

Так как задача одновременно IO-bound (получение записей по http) и CPU-bound (парсинг сертификатов), для эффективной обработки необходимо будет подключить как асинхронность, так и многопроцессность.

Так как не было никаких инструментов, которые позволили бы легко и безболезненно получить и распарсить все CTL (помимо не особо примечательной утилиты от Google, было решено потратить немного времени и написать свой инструмент, который соответствовал бы всем нашим потребностям. Результатом стал Axeman, который использует asyncio и замечательную библиотеку aioprocessing для загрузки, парсинга и сохранения сертификатов в несколько CSV файлов, ограничиваясь при этом только скоростью интернет-соединения.

Эксплуатация облака

После получения инстанса (_прим. перев._ так в Google Cloud называются VM) c 16 ядрами, 32Гб памяти и SSD на 750Гб (спасибо Google за бесплатные 300$ на счете для новых аккаунтов!), я запустил Axeman, который загрузил все сертификаты меньше чем за сутки и сохранил результаты в /tmp/certificates/$CTL_DOMAIN/

Где хранить все эти данные?

На начальном этапе для осуществления хранения и поиска по данным был выбран Postgres, но, хотя я и не сомневаюсь, что с правильной схемой Postgres легко бы справился с 250 миллионами записей (в отличии от моей первой попытки, в которой на один запрос уходило примерно 20 минут!), я начал искать решения, которые:

• позволяют дешево хранить большой объем данных

• обеспечивают быстрый поиск

• позволяют легко обновлять данные

Вариантов было несколько, но с точки зрения стоимости, почти все рассмотренные варианты (AWS RDS, Heroku Postgres, Google Cloud SQL) были весьма затратны. К счастью, так как наши данные в принципе никогда не изменяются, у нас появляется дополнительная гибкость в выборе платформы для размещения данных.

В целом, это как раз тот тип поиска по данным, который прекрасно ложится на модель map/reduce с использованием, к примеру, Spark или Hadoop Pig. Просматривая предложения различных провайдеров в категории “big data” (хотя в нашей задаче данных явно мало для включения в эту категорию), я наткнулся на Google BigQuery, который удовлетворяет всем обозначенным параметрам.

Скармливаем данные BigQuery

Загрузка данных в BigQuery осуществляется довольно легко, благодаря предоставляемой Google утилите gsutil. Создаем новый бакет для наших сертификатов:

Когда бакет готов, используем gsutil для транспортировки всех сертификатов в хранилище Google (а затем в BigQuery). После настройки аккаунта командой gsutil config, запускаем процесс загрузки:

gsutil -o GSUtil:parallel_composite_upload_threshold=150M        -m cp        /tmp/certificates/*        gs://all-certificates

И видим следующий результат в нашем бакете:

Далее создаем новый датасет в BigQuery:

Теперь мы можем импортировать данные из хранилища в наш новый датасет. К сожалению, в BigQuery нет кнопки “пожалуйста, импортируй все папки рекурсивно”, так что придется импортировать каждый CTL отдельно, но занимает это не так долго. Создаем таблицу и импортируем наш первый лог (обратите особое внимание на отмеченные настройки):

Так как схема нужна всякий раз при импорте очередного лога, воспользуемся опцией “Edit as Text”. Использованная схема:

[
    {
        "name": "url",
        "type": "STRING",
        "mode": "REQUIRED"
    },
    {
        "mode": "REQUIRED",
        "name": "cert_index",
        "type": "INTEGER"
    },
    {
        "mode": "REQUIRED",
        "name": "chain_hash",
        "type": "STRING"
    },
    {
        "mode": "REQUIRED",
        "name": "cert_der",
        "type": "STRING"
    },
    {
        "mode": "REQUIRED",
        "name": "all_dns_names",
        "type": "STRING"
    },
    {
        "mode": "REQUIRED",
        "name": "not_before",
        "type": "FLOAT"
    },
    {
        "mode": "REQUIRED",
        "name": "not_after",
        "type": "FLOAT"
    }
]

Далее просто повторяем процесс для каждого лога. Убедитесь, что каждый импорт завершился успешно (ошибки обычно можно игнорировать, просто убедитесь, что вы задали адекватное значения для максимального количества ошибок). В итоге должен получиться примерно такой датасет:

Что в итоге получилось

Настало время пожинать плоды наших трудов и опробовать систему на различных запросах.

В последнее время часто говорят о доменах, использующих punycode и связанных с ними омоглифных атаках. Попробуем следующий запрос:

SELECT
  all_dns_names
FROM
  [ctl-lists:certificate_data.scan_data]
WHERE
  (REGEXP_MATCH(all_dns_names,r'\b?xn\-\-'))
  AND NOT all_dns_names CONTAINS 'cloudflare'

И всего через 15 секунд получаем результат со всеми punycode доменами из всех известных CTL!

Рассмотрим другой пример. Попробуем получить все сертификаты доменов Coinbase, записанные в Certificate Transparency:

SELECT
  all_dns_names
FROM
  [ctl-lists:certificate_data.scan_data]
WHERE
  (REGEXP_MATCH(all_dns_names,r'.*\.coinbase.com[\s$]?'))

Всего через две секунды получаем все интересующие нас результаты:

Возможность легко проводить подобный анализ на таком обширном датасете - сильный инструмент для идентификации различных трендов, которые невозможно было бы обнаружить каким-либо другим способом.

Небольшая загадка

Проводя исследования, я обнаружил нечто странное. Домен flowers-to-the-world.com постоянно возникал в различных логах. Практически каждый лог имел огромное число сертификатов, содержащих этот домен:

SELECT
  url,
  COUNT(*) AS total_certs
FROM
  [ctl-lists:certificate_data.scan_data]
WHERE
  (REGEXP_MATCH(all_dns_names,r'.*flowers-to-the-world.*'))
GROUP BY
  url
ORDER BY
  total_certs DESC

Whois позволяет определить, что этот домен принадлежит Google, так что мне интересно, не является ли это частью какой-то тестовой рутины. Если вы инженер Google, который может разузнать что-то у товарищей, занимающихся Certificate Transparency, было бы очень интересно услышать об этом.

Совсем скоро мы планируем выпустить собственный продукт — NetLas.io. Это своеобразный технический атлас всей сети Интернет, который будет включать не только сертификаты, но и данные по доменам и поддоменам, ответы серверов по популярным портам и много другой полезной для исследователей безопасности информации.

В России, насколько нам известно, это первый такой продукт. В США и Китае у нас есть сильные конкуренты, но мы надеемся превзойти их по некоторым параметрам. Например, актуальность данных — уже сейчас наша реализация поискового движка позволяет включать в поисковую выдачу данные, от сканов не старше минуты. На сегодняшний день Netlas.io доступен в формате "раннего доступа". Если есть желание потестировать — переходите на сайт и регистрируйтесь.