Opensource Mail Gateway ürünü olan proxmox mailgateway’in(PMG) kendi içerisinde bir çok filtreleme katmanı bulunuyor. Bunlara ek olarak bir çok ücretli mail gateway yazılımında dahi bulunmayan bir entegrasyonu hep birlikte yapalım. Bunun için PMG’nin Custom Check Interface özelliğinden faydalanacağiz.

Öncelikle;

virustotal üyeliği açarak api yönlendirmelerini takip edip üyeliğimize ait bir api key almamız gerekmektedir. virustotal’e sorgu atarken bu api keyden faydalanacağız.

Önemli Not: virustotal free üyelikde günde 500 sorguya kadar izin vermektedir. premium fiyatları dolar bazlı olduğundan ve astarı yüzünü geçtiğinden şöyle bir çözüm ürettim.

bir url’i virustotal’e sorguladıktan sonra sonucunu proxmox içerisinde yer alan postgres db’ye yazdım. ve aynı url tekrar geldiğinde son 7 gün içerisinde sorgulamışsam, tekrar virustotal’den sorgulamadım yerel db’imde yer alan sonuçları referans aldım.

/etc/pmg/pmg.conf dosyasında admin bölümünde custom check’i enable edelim.

section: admin
    custom_check 1

Custom check interface bash scriptini default ayarlarda “/usr/local/bin/pmg-custom-check” path’ine yazmamız gerekiyor. İsterseniz bu path’i de değiştirebilirsiniz. Burada oluşturduğumuz bir bash script ile asıl işi yapacak olan python dosyasını çağıracağız.

#!/bin/bash

parametre1=$1
parametre2=$2

python_sonuc=$(python3 /home/pmg-custom-check.py $parametre1 $parametre2)


echo "v1"
echo "$python_sonuc"


exit 0

Bash script dosyamız ile /home/pmg-custom-check.py yolundaki python dosyamızı çağırıyoruz. “echo "$python_sonuc" satırında python dosyamızdan aldığımız yanıtı proxmox’a geri gönderiyoruz.

python dosyamız;

import sys
from urlextract import URLExtract
import re
import json
import requests
import base64
import tldextract
import psycopg2
import datetime


# Dışarıdan gelen parametreleri al ve mail dosyasını oku
api_version = sys.argv[1] if len(sys.argv) > 1 else None
current_mail = sys.argv[2] if len(sys.argv) > 2 else None
mail_icerik_file = open(current_mail, "r")
mail_icerik = mail_icerik_file.read()

## tanımlamalar

virus_total_api_key = "xxxxxxxxxxxxxxxx"
guvenilen_domainler = ["microsoft.com","w3.org"]
kontrol_edilecek_domainler = []
kontrol_edilecek_subdomainler = []
kontrol_edilecek_urller = []
base64_anahtar_kelime = "Content-Transfer-Encoding: base64"
base64_count = mail_icerik.count(base64_anahtar_kelime)

#db tanımlamaları

db_user = "virustotallocal_db_user"
db_password = "xxxxxx"
db_host = "127.0.0.1"
db_name = "virustotallocal_db"

## fonksiyonlar

def db_select(tur,data):
    today = datetime.date.today()
    connection = psycopg2.connect(user=db_user,
                                  password=db_password,
                                  host=db_host,
                                  port="5432",
                                  database=db_name)
    cursor = connection.cursor()
    sorgu = """select "malicious","suspicious" from sorgu_sonuclari
    WHERE tur = %s  and data = %s and gecerlilik_tarihi > %s"""
    cursor.execute((sorgu),(tur,data,today))
    sorgu_sonuc = cursor.fetchall()
    return sorgu_sonuc


def db_insert(tur,data,malicious,suspicious):
    otuzgun = datetime.date.today() + datetime.timedelta(30)
    connection = psycopg2.connect(user=db_user,
                                  password=db_password,
                                  host=db_host,
                                  port="5432",
                                  database=db_name)
    connection.autocommit = True
    cursor = connection.cursor()
    sqlquery =  """INSERT INTO sorgu_sonuclari(tur, data, malicious, suspicious, gecerlilik_tarihi, created_date) VALUES (%s, %s, %s, %s, %s, %s)"""
    cursor.execute(sqlquery,(tur,data,malicious,suspicious,otuzgun, datetime.datetime.now()))
    connection.commit()
    connection.close()



def url_id_bul(url):
    virus_total_sorgu_url = "https://www.virustotal.com/api/v3/urls"
    payload = { "url": url }
    headers = {
        "accept": "application/json",
        "x-apikey": virus_total_api_key,
        "content-type": "application/x-www-form-urlencoded"
    }
    response = requests.post(virus_total_sorgu_url, data=payload, headers=headers)
    response_json = json.loads(response.text)
    id = response_json['data']['id']
    return id

def id_sonuc_bul(id):
    virus_total_sorgu_url = "https://www.virustotal.com/api/v3/analyses/{}".format(id)
    headers = {
        "accept": "application/json",
        "x-apikey": virus_total_api_key
    }
    response = requests.get(virus_total_sorgu_url, headers=headers)   
    sonuc = json.loads(response.text)
    stats = sonuc['data']['attributes']['stats']
    return stats



def mail_icerik_find_urls(mail_icerik):
    ## base64 maili string'e çevir
    if base64_count > 0:
        search_baslangici = 0
        for x in range(base64_count):
                try:
                    baslangic = mail_icerik.find(base64_anahtar_kelime, search_baslangici)
                    bosluk_index = mail_icerik.find("--", baslangic+len(base64_anahtar_kelime))
                    search_baslangici = bosluk_index
                    sonuc = mail_icerik[baslangic+len(base64_anahtar_kelime):bosluk_index].strip()
                    base64_message = sonuc
                    base64_bytes = base64_message.encode('ascii')
                    message_bytes = base64.b64decode(base64_bytes)
                    message = message_bytes.decode('utf-8')
                    mail_icerik += message
                except:
                    pass
        mail_icerik = mail_icerik.replace('\n', '').replace('=', '').replace('--', ' ').replace('</a>', ' ').replace('<', ' ').replace('>', ' ').replace(']', ' ')
        urls = re.findall('http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', mail_icerik)
    else:
        mail_icerik = mail_icerik.replace('\n', '').replace('=', '').replace('--', ' ').replace('</a>', ' ')
        urls = re.findall('http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', mail_icerik)
    return urls


def url_to_domain(url): #verilen url'nin domainini ayrıştırır ve güvenilir domainler listesinde değilse kontrol edilecek domainler listesine ekler.
    ayristirilmis_url = tldextract.extract(url)
    domain = ayristirilmis_url.domain+'.'+ayristirilmis_url.suffix
    if domain in guvenilen_domainler or domain in kontrol_edilecek_domainler:
        pass
    else:
        kontrol_edilecek_domainler.append(domain)

def url_to_subdomain(url): #verilen url'nin subdomainini ayrıştırır ve güvenilir subdomainler listesinde değilse kontrol edilecek subdomainler listesine ekler.
    ayristirilmis_url = tldextract.extract(url)
    domain = ayristirilmis_url.domain+'.'+ayristirilmis_url.suffix
    subdomain = ayristirilmis_url.subdomain+'.'+ayristirilmis_url.domain+'.'+ayristirilmis_url.suffix
    if domain in guvenilen_domainler or subdomain in kontrol_edilecek_subdomainler:
        pass
    else:
        kontrol_edilecek_subdomainler.append(subdomain)

def url_append_kontrol_edilecek(url): #verilen url'nin domaini güvenilir domainler listesinde var mı kontrol eder, yoksa kontrol edilecek urller listesine ekler
    ayristirilmis_url = tldextract.extract(url)
    domain = ayristirilmis_url.domain+'.'+ayristirilmis_url.suffix
    if domain in guvenilen_domainler or url in kontrol_edilecek_urller:
        pass
    else:
        kontrol_edilecek_urller.append(url)

def spam_score_hesapla(malicious,suspicious):
    skor = float(malicious) + (float(suspicious * 0.50))
    return skor



## mail içerisindeki url'leri bulma
urls = mail_icerik_find_urls(mail_icerik)
urls = list(dict.fromkeys(urls))
for url in urls:
    url_append_kontrol_edilecek(url)
    url_to_domain(url)
    url_to_subdomain(url)



def domainler_spam_score(kontrol_edilecek_domainler):
    skor = 0
    for domain in kontrol_edilecek_domainler:
        domain_select = db_select("domain",domain)
        if len(domain_select) > 0:
            domain_malicious = domain_select[0][0]
            domain_suspicious = domain_select[0][1]
            skor += spam_score_hesapla(domain_malicious,domain_suspicious)
        else:
            domain_id=url_id_bul(domain)
            domain_sonuc = id_sonuc_bul(domain_id)
            domain_malicious = domain_sonuc["malicious"]
            domain_suspicious = domain_sonuc["suspicious"]
            skor += spam_score_hesapla(domain_malicious,domain_suspicious)
            db_insert("domain",domain,domain_malicious,domain_suspicious)
    return skor



def sub_domainler_spam_score(kontrol_edilecek_subdomainler):
    skor = 0
    for subdomain in kontrol_edilecek_subdomainler:
        subdomain_select = db_select("subdomain",subdomain)
        if len(subdomain_select) > 0:
            subdomain_malicious = subdomain_select[0][0]
            subdomain_suspicious = subdomain_select[0][1]
            skor += spam_score_hesapla(subdomain_malicious,subdomain_suspicious)
        else:
            subdomain_id=url_id_bul(subdomain)
            subdomain_sonuc = id_sonuc_bul(subdomain_id)
            subdomain_malicious = subdomain_sonuc["malicious"]
            subdomain_suspicious = subdomain_sonuc["suspicious"]
            skor += spam_score_hesapla(subdomain_malicious,subdomain_suspicious)
            db_insert("subdomain",subdomain,subdomain_malicious,subdomain_suspicious)
    return skor

def sub_domainler_spam_score(kontrol_edilecek_subdomainler):
    skor = 0
    for subdomain in kontrol_edilecek_subdomainler:
        subdomain_select = db_select("subdomain",subdomain)
        if len(subdomain_select) > 0:
            subdomain_malicious = subdomain_select[0][0]
            subdomain_suspicious = subdomain_select[0][1]
            skor += spam_score_hesapla(subdomain_malicious,subdomain_suspicious)
        else:
            subdomain_id=url_id_bul(subdomain)
            subdomain_sonuc = id_sonuc_bul(subdomain_id)
            subdomain_malicious = subdomain_sonuc["malicious"]
            subdomain_suspicious = subdomain_sonuc["suspicious"]
            skor += spam_score_hesapla(subdomain_malicious,subdomain_suspicious)
            db_insert("subdomain",subdomain,subdomain_malicious,subdomain_suspicious)
    return skor


def urller_spam_score(kontrol_edilecek_subdomainler):
    skor = 0
    for url in kontrol_edilecek_urller:
        url_select = db_select("url",url)
        if len(url_select) > 0:
            url_malicious = url_select[0][0]
            url_suspicious = url_select[0][1]
            skor += spam_score_hesapla(url_malicious,url_suspicious)
        else:
            url_id=url_id_bul(url)
            url_sonuc = id_sonuc_bul(url_id)
            url_malicious = url_sonuc["malicious"]
            url_suspicious = url_sonuc["suspicious"]
            skor += spam_score_hesapla(url_malicious,url_suspicious)
            db_insert("url",url,url_malicious,url_suspicious)
    return skor


domainler_spam_score(kontrol_edilecek_domainler)
sub_domainler_spam_score(kontrol_edilecek_subdomainler)
urller_spam_score(kontrol_edilecek_urller)

mail_spam_score = domainler_spam_score(kontrol_edilecek_domainler) + sub_domainler_spam_score(kontrol_edilecek_subdomainler) + urller_spam_score(kontrol_edilecek_urller)


if mail_spam_score > 1:
    print("SCORE: 10")
else:
    print("OK")