Cara Menemukan Army Dari Fake User Di Python - CRUDPRO

Cara Menemukan Army Dari Fake User Di Python

Cara Menemukan Army Dari Fake User Di Python

Mengurangi serangan rantai pasokan perangkat lunak adalah permainan kucing dan tikus yang tidak ada habisnya. Kami, sebagai pembela, selalu berusaha menemukan cara untuk menutup jarak antara waktu dimulainya serangan dan saat kami dapat menghentikannya atau setidaknya melindungi calon korban.

Salah satu cara melakukannya adalah mengantisipasi gerakan penyerang dan berada di sana saat mereka melakukannya. Hal ini dimungkinkan ketika pembela HAM dapat mengidentifikasi pola dalam tindakan penyerang berdasarkan gerakan mereka dan "memburu" mereka. Kadang-kadang kita dapat melakukannya dengan menggunakan data yang sudah kita miliki. Dalam kasus lain, data ini harus diperoleh terlebih dahulu, seperti kejadian kami baru-baru ini.

Pendahuluan

Akhir pekan lalu, kami melaporkan serangan besar-besaran pada ekosistem Python. Lebih dari 400 paket jahat diterbitkan oleh 22+ akun pengguna terpisah yang bertujuan untuk mencuri dompet kripto dari pengguna Python.

Seiring berjalannya minggu, jumlah paket berbahaya yang diterbitkan oleh penyerang terus meningkat, melebihi 900 paket dari puluhan akun pengguna. Kami semua menyaksikan serangan yang sedang berlangsung tanpa petunjuk kapan akan berakhir.

Kami perlu mengembangkan strategi proaktif untuk tetap berada di depan penyerang dan mencegah mereka membahayakan korban lagi. Meskipun kami telah memantau akun pengguna yang telah menerbitkan paket berbahaya, kami juga khawatir tentang akun pengguna tidak dikenal yang terkait dengan penyerang tetapi belum merilis paket berbahaya apa pun.

Untuk mengatasi masalah ini, saya menyelidiki lebih lanjut dan menemukan bahwa semua akun pengguna yang terkait dengan serangan yang sedang berlangsung dibuat pada hari yang sama, tanggal 4 November. Saya kemudian membuat peta dari semua akun pengguna yang mencurigakan yang dibuat di platform PyPI pada tanggal 4 November dan belum merilis paket apa pun (ada banyak paket!). Dengan memantau akun-akun ini untuk aktivitas apa pun, saya dapat menangkap paket berbahaya saat dirilis, yang secara efektif menggagalkan upaya penyerang.

Di blog ini, saya akan membagikan strategi saya untuk mengidentifikasi dan memantau akun pengguna yang mencurigakan untuk mencegah serangan di masa depan.

Dapatkan Semua Akun Pengguna PyPi

Tujuan pertama saya adalah API PyPi, karena API tersebut menunjukkan jumlah total akun pengguna di halaman utama.

Cara Menemukan Army Dari Fake User Di Python

PyPi memang menyediakan berbagai API REST untuk mengakses informasi tentang paket yang dihosting di platformnya. Beberapa contohnya adalah:

https://pypi.org/pypi/{package_name}/json

Yang mengembalikan informasi terperinci tentang paket tertentu, termasuk versi dan metadata paket.

Ada juga XML-RPC API dari PyPi yang menyediakan berbagai metode untuk berinteraksi dengan platformnya, seperti mencari paket dan mengambil metadata paket.

Sayangnya, tidak ada API PyPi yang mendukung mendapatkan daftar semua akun pengguna yang dipublikasikan dalam jangka waktu tertentu, maupun daftar lengkap semua pengguna secara umum.

sitemap.xml Untuk Penyelamatan

Dalam SEO, peta situs XML digunakan untuk membuat daftar halaman-halaman penting situs web, memastikan mesin telusur seperti Google dapat menemukan dan merayapi semuanya.

Seperti yang saya duga, file sitemap.xml PyPi berisi daftar halaman yang terkait dengan semua akun pengguna, paket, dan lainnya.

Cara Menemukan Army Dari Fake User Di Python

Untuk mengekstrak daftar akun pengguna dari sumber ini, saya membuat pengikis sederhana di Python dengan bantuan perpustakaan bernama beautifulsoup. Ini melewati setiap sub-halaman dari file peta situs dan mengekstrak semua URL akun pengguna di situs web PyPi, secara efektif memberi kami daftar semua pengguna di registri ini.

def get_all_pypi_usernames(session):
    url = "https://pypi.org/sitemap.xml"
    r = session.get(url)
    r.raise_for_status()
    soup = BeautifulSoup(r.text, 'xml')
    sitemaps = soup.findAll('sitemap')
    sitemap_urls = map(lambda x: x.find('loc').text, sitemaps)
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(session.get, sitemap_url) for sitemap_url in sitemap_urls]
        for future in concurrent.futures.as_completed(futures):
            r = future.result()
            r.raise_for_status()
            soup = BeautifulSoup(r.text, 'xml')
            url_elements = soup.findAll('url')
            pypi_urls = map(lambda x: x.find('loc').text, url_elements)
            pypi_urls = filter(lambda x: x.startswith('https://pypi.org/user/'), pypi_urls)
            pypi_urls = map(lambda x: x.replace('https://pypi.org/user/', '').strip('/'), pypi_urls)
            pypi_urls = map(lambda x: x.lower(), pypi_urls)
            yield from pypi_urls

Mendapatkan Lebih Banyak Data Pengguna

Setelah mendapatkan daftar lengkap lebih dari 620 ribu akun pengguna, saya ingin mendapatkan lebih banyak data untuk setiap pengguna.

Saya memperluas scraper saya untuk memuat halaman setiap pengguna di PyPi menggunakan pola URL

https://pypi.org/user/<username>
dan mengekstraksi data menarik seperti:
  • Nama tampilan
  • tanggal registrasi
  • Jumlah paket
  • URL gambar profil (lebih lanjut tentang itu sebentar lagi)

Meskipun PyPi menampilkan hari akun pengguna dibuat, saya dapat mengambil stempel waktu setelah memeriksa kode sumber halaman dan mengekstrak stempel waktu dari atribut datetime.

Cara Menemukan Army Dari Fake User Di Python

Patut dicatat bahwa tidak semua halaman profil pengguna menyediakan tanggal pendaftaran. Hipotesis saya adalah PyPi mulai mengaudit bidang ini untuk pengguna yang dibuat setelah 2013, dan ini hilang untuk akun pengguna yang terdaftar sebelum 17 September 2013.

def get_pypi_user_info(username, session):
    url = f'https://pypi.org/user/{username}/'
    r = session.get(url)
    r.raise_for_status()
    soup = BeautifulSoup(r.content, 'html.parser')
    profile_info_element = soup.find(class_='author-profile__info')
    registration_date = ''
    author_profile = soup.find(class_='author-profile')
    profile_image_url = author_profile.find('img')['src']

    r = session.get(profile_image_url)
    r.raise_for_status()

    image = Image.open(BytesIO(r.content))
    is_gravatar_default_image = list(image.getdata()) == list(DEFAULT_GRAVATAR_IMAGE.getdata())

    if profile_image_url and profile_image_url.startswith('https://warehouse-camo.ingress.cmh1.psfhosted.org'):
        hex_encoded_url = profile_image_url.split('/')[-1]
        profile_image_url = bytes.fromhex(hex_encoded_url).decode()

    registration_date_element = profile_info_element.find('time')
    if registration_date_element:
        registration_date = registration_date_element.attrs.get('datetime', '')

    display_name = ''
    profile_name_element = soup.find(class_='author-profile__name')
    if profile_name_element:
        display_name = profile_name_element.text

    published_packages = soup.findAll(class_='package-snippet')
    published_packages_count = len(published_packages)

    return display_name, registration_date, published_packages_count, profile_image_url, is_gravatar_default_image

Memeriksa apakah pengguna memiliki gambar default

Untuk memeriksa apakah akun pengguna mengubah gambar profil (melalui Gravatar), saya menulis kode berikut untuk membandingkan gambar pengguna dengan gambar default gravatar. Ini bekerja dengan sangat baik.

// download this image from:
// https://secure.gravatar.com/avatar/default?size=225
DEFAULT_GRAVATAR_IMAGE = Image.open('gravatar_template_image.jpg')

.
.
.
.

r = session.get(profile_image_url)
r.raise_for_status()

image = Image.open(BytesIO(r.content))
is_gravatar_default_image = list(image.getdata()) == list(DEFAULT_GRAVATAR_IMA

Hasil

Skrip lengkap yang saya buat ada di sini.

import concurrent
import csv
import logging
from concurrent.futures import ThreadPoolExecutor
from io import BytesIO

import requests
from bs4 import BeautifulSoup
from requests.adapters import HTTPAdapter
from tqdm import tqdm
from urllib3 import Retry
from PIL import Image

// download this image from:
// https://secure.gravatar.com/avatar/default?size=225
DEFAULT_GRAVATAR_IMAGE = Image.open('gravatar_template_image.jpg')


def get_pypi_user_info(username, session):
    url = f'https://pypi.org/user/{username}/'
    r = session.get(url)
    r.raise_for_status()
    soup = BeautifulSoup(r.content, 'html.parser')
    profile_info_element = soup.find(class_='author-profile__info')
    registration_date = ''
    author_profile = soup.find(class_='author-profile')
    profile_image_url = author_profile.find('img')['src']

    r = session.get(profile_image_url)
    r.raise_for_status()

    image = Image.open(BytesIO(r.content))
    is_gravatar_default_image = list(image.getdata()) == list(DEFAULT_GRAVATAR_IMAGE.getdata())

    if profile_image_url and profile_image_url.startswith('https://warehouse-camo.ingress.cmh1.psfhosted.org'):
        hex_encoded_url = profile_image_url.split('/')[-1]
        profile_image_url = bytes.fromhex(hex_encoded_url).decode()

    registration_date_element = profile_info_element.find('time')
    if registration_date_element:
        registration_date = registration_date_element.attrs.get('datetime', '')

    display_name = ''
    profile_name_element = soup.find(class_='author-profile__name')
    if profile_name_element:
        display_name = profile_name_element.text

    published_packages = soup.findAll(class_='package-snippet')
    published_packages_count = len(published_packages)

    return display_name, registration_date, published_packages_count, profile_image_url, is_gravatar_default_image


def get_all_pypi_usernames(session):
    url = "https://pypi.org/sitemap.xml"
    r = session.get(url)
    r.raise_for_status()
    soup = BeautifulSoup(r.text, 'xml')
    sitemaps = soup.findAll('sitemap')
    sitemap_urls = map(lambda x: x.find('loc').text, sitemaps)
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(session.get, sitemap_url) for sitemap_url in sitemap_urls]
        for future in concurrent.futures.as_completed(futures):
            r = future.result()
            r.raise_for_status()
            soup = BeautifulSoup(r.text, 'xml')
            url_elements = soup.findAll('url')
            pypi_urls = map(lambda x: x.find('loc').text, url_elements)
            pypi_urls = filter(lambda x: x.startswith('https://pypi.org/user/'), pypi_urls)
            pypi_urls = map(lambda x: x.replace('https://pypi.org/user/', '').strip('/'), pypi_urls)
            pypi_urls = map(lambda x: x.lower(), pypi_urls)
            yield from pypi_urls


def main():
    session = requests.session()
    retry = Retry(connect=3, backoff_factor=0.5)
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s - %(message)s')

    logging.info('loading usernames...')
    pypi_usernames = get_all_pypi_usernames(session)
    pypi_usernames = list(pypi_usernames)
    pypi_usernames = pypi_usernames

    logging.info(f'loaded {len(pypi_usernames)} usernames')

    with open('users.csv', 'w+') as f, tqdm(total=len(pypi_usernames)) as progress:
        writer = csv.writer(f)
        writer.writerow(['username', 'display name', 'registration date', 'number of packages', 'image url', 'is default image'])

        with ThreadPoolExecutor(max_workers=10) as executor:
            futures_map = {executor.submit(get_pypi_user_info, pypi_username, session): pypi_username for pypi_username in pypi_usernames}
            for future in concurrent.futures.as_completed(futures_map):
                pypi_username = futures_map[future]
                try:
                    display_name, registration_date, number_of_packages, image_url, is_gravatar_default_image = future.result()
                    writer.writerow([pypi_username, display_name, registration_date, number_of_packages, image_url, is_gravatar_default_image])
                    progress.update()
                    f.flush()
                except:
                    logging.exception(f'failed to process {pypi_username}')


if __name__ == '__main__':
    main()

Menemukan Penyimpangan

Sesudah mengekstrak info ini berkenaan akun pemakai di PyPi, saya mulai cari penyimpangan, dan lebih khusus kembali — akun pemakai yang berkaitan dengan kejadian minggu lalu, di mana seorang mengeluarkan 900 paket kesalahan tulis berbahaya menggunakan lebih dari 30 akun pengguna yang lain. Semuanya dibuat sebelumnya beberapa waktu awalnya, pada 4 November 2022.

Cara Menemukan Army Dari Fake User Di Python
Cara Menemukan Army Dari Fake User Di Python

Saya membuat daftar akun pengguna PyPi yang mencurigakan, yang menurut saya mungkin terkait dengan serangan kesalahan ketik dari minggu lalu. Berbagi daftar di bawah ini:

No
Username
Display Name
Registration Date
1 ecnldear07 ecnldear07 2022-11-04T18:14:27+0000
2 peke-0393 peke-0393 2022-11-04T18:14:13+0000
3 klenin281 klenin281 2022-11-04T18:13:57+0000
4 makova1994 makova1994 2022-11-04T18:13:39+0000
5 pinigin.9494 pinigin.9494 2022-11-04T18:13:15+0000

Ringkasan

Upaya saya menghasilkan daftar ~280 akun pengguna PyPi yang diduga menjadi bagian dari infrastruktur penyerang yang dibuat pada 4 November 2022, untuk penggunaan di masa mendatang.

Saya meminta pembela lain untuk memantau daftar akun pengguna yang mencurigakan ini kalau-kalau akun tersebut diaktifkan dan mulai menerbitkan paket berbahaya baru.

Menyadari bahwa ini bukan masalah paket jahat tetapi masalah penyerang, mengubah cara berpikir kita dan mungkin membantu kita mencegah serangan lebih dekat dengan penciptaannya.