Add files via upload

PYAS.py

@@ -1,16 +1,13 @@
 import os, gc, sys, time, json, psutil
 import hashlib, pefile, socket, msvcrt
+import xml.etree.ElementTree as xmlet
 import requests, pyperclip, win32file
 import win32gui, win32api, win32con
-import xml.etree.ElementTree as xmlet
-from PYAS_Database import model_dict
-from PYAS_Function_Virus import func_virus
-from PYAS_Function_Safe import func_safe
 from PYAS_Extension import slist, alist
+from PYAS_Function_Safe import func_safe
+from PYAS_Function_Virus import func_virus
 from PYAS_Language import translate_dict
 from PYAS_Interface import Ui_MainWindow
-from PYML import ListClassifier
-from binascii import hexlify
 from threading import Thread
 from PyQt5.QtWidgets import *
 from PyQt5.QtCore import *
@@ -22,11 +19,10 @@ class MainWindow_Controller(QMainWindow):
         self.setAttribute(Qt.WA_TranslucentBackground)
         self.setWindowFlags(Qt.FramelessWindowHint)
         self.pyas = str(sys.argv[0]).replace("\\", "/")
-        self.pyas_version = "2.9.7"
+        self.pyas_version = "2.9.8"
         self.ui = Ui_MainWindow()
         self.ui.setupUi(self)
         self.init_tray_icon()
-        self.init_data_base()
         self.init_virus_list()
         self.init_state_safe()
         self.init_config_path()
@@ -40,10 +36,6 @@ class MainWindow_Controller(QMainWindow):
         self.show_pyas_ui()
         self.init_threads()
 
-    def init_data_base(self):
-        self.ml = ListClassifier()
-        self.ml.load_model(model_dict)
-
     def init_threads(self):
         self.protect_proc_init()
         self.protect_file_init()
@@ -277,7 +269,7 @@ class MainWindow_Controller(QMainWindow):
         self.ui.Window_Block_Button.setText(self.trans("軟體彈窗攔截"))
         self.ui.Repair_System_Network_Button.setText(self.trans("系統網路修復"))
         self.ui.About_Back.setText(self.trans("返回"))
-        self.ui.PYAS_Version.setText(self.trans(f"PYAS V{self.pyas_version} (ML Engine Beta)"))
+        self.ui.PYAS_Version.setText(self.trans(f"PYAS V{self.pyas_version}"))
         self.ui.GUI_Made_title.setText(self.trans("介面製作:"))
         self.ui.GUI_Made_Name.setText(self.trans("mtkiao"))
         self.ui.Core_Made_title.setText(self.trans("核心製作:"))
@@ -558,11 +550,11 @@ class MainWindow_Controller(QMainWindow):
         except:
             pass
 
-    def send_notify(self,text):
+    def send_notify(self, text):
         try:
             now_time = time.strftime('%Y/%m/%d %H:%M:%S')
-            self.ui.State_output.append(f"[{now_time}] {text}")
             self.tray_icon.showMessage(now_time, text, 5000)
+            QMetaObject.invokeMethod(self.ui.State_output, "append", Qt.QueuedConnection, Q_ARG(str, f"[{now_time}] {text}"))
         except:
             pass
 
@@ -786,18 +778,14 @@ class MainWindow_Controller(QMainWindow):
             file_type = str(f".{file.split('.')[-1]}").lower()
             if self.high_sensitivity and file != self.pyas:
                 if self.api_scan(file):
-                    self.write_scan(self.trans("Vir.cloud.md5"),file)
-                elif self.ml_scan(file):
-                    self.write_scan(self.trans("Sus.hdfile.ml"),file)
+                    self.write_scan(self.trans("惡意"),file)
                 elif self.pe_scan(file):
-                    self.write_scan(self.trans("Sus.pefile.he"),file)
+                    self.write_scan(self.trans("可疑"),file)
             elif file_type in slist and file != self.pyas:
                 if self.api_scan(file):
-                    self.write_scan(self.trans("Vir.cloud.md5"),file)
-                elif self.ml_scan(file):
-                    self.write_scan(self.trans("Sus.hdfile.ml"),file)
+                    self.write_scan(self.trans("惡意"),file)
                 elif self.pe_scan(file):
-                    self.write_scan(self.trans("Sus.pefile.he"),file)
+                    self.write_scan(self.trans("可疑"),file)
         except:
             pass
 
@@ -824,6 +812,7 @@ class MainWindow_Controller(QMainWindow):
             if self.cloud_services == 1:
                 with open(file, "rb") as f:
                     text = str(hashlib.md5(f.read()).hexdigest())
+                QApplication.processEvents()
                 strBody = f'-------------------------------7d83e2d7a141e\r\nContent-Disposition: form-data; name="md5s"\r\n\r\n{text}\r\n-------------------------------7d83e2d7a141e\r\nContent-Disposition: form-data; name="format"\r\n\r\nXML\r\n-------------------------------7d83e2d7a141e\r\nContent-Disposition: form-data; name="product"\r\n\r\n360zip\r\n-------------------------------7d83e2d7a141e\r\nContent-Disposition: form-data; name="combo"\r\n\r\n360zip_main\r\n-------------------------------7d83e2d7a141e\r\nContent-Disposition: form-data; name="v"\r\n\r\n2\r\n-------------------------------7d83e2d7a141e\r\nContent-Disposition: form-data; name="osver"\r\n\r\n5.1\r\n-------------------------------7d83e2d7a141e\r\nContent-Disposition: form-data; name="vk"\r\n\r\na03bc211\r\n-------------------------------7d83e2d7a141e\r\nContent-Disposition: form-data; name="mid"\r\n\r\n8a40d9eff408a78fe9ec10a0e7e60f62\r\n-------------------------------7d83e2d7a141e--'
                 response = requests.post('http://qup.f.360.cn/file_health_info.php', data=strBody, timeout=3)
                 return response.status_code == 200 and float(xmlet.fromstring(response.text).find('.//e_level').text) > 50
@@ -858,38 +847,17 @@ class MainWindow_Controller(QMainWindow):
                 max_sfl.append(len(set(fn)&set(sfl))/len(set(fn)|set(sfl)))
             if self.high_sensitivity:
                 return max(max_vfl) >= max(max_sfl)
-            elif "_CorExeMain" not in fn:
-                return max(max_vfl) >= max(max_sfl)
-            return False
+            return max(max_vfl) - max(max_sfl) >= 0.1
         except:
             return False
 
-    def ml_scan(self, file):
+    def proc_scan(self, pid):
         try:
-            data = []
-            if os.path.getsize(file) <= 209715200:
-                with open(file, 'rb') as f:
-                    for block in iter(lambda: f.read(3),b''):
-                        QApplication.processEvents()
-                        data.append(str(hexlify(block),'utf-8'))
-            if self.high_sensitivity:
-                return self.ml.predict(data) == 1
-            elif self.sign_scan(file):
-                return self.ml.predict(data) == 1
-            return False
-        except:
-            return False
-
-    def scan_proc(self, value):
-        try:
-            if isinstance(value, int):
-                for entry in psutil.Process(value).memory_maps():
-                    file = entry.path.replace("\\", "/")
-                    if ":/Windows" not in file and ":/Program" not in file and "/AppData/" not in file:
-                        if self.api_scan(file) or self.pe_scan(file) or self.ml_scan(file):
-                            return True
-            elif self.api_scan(value) or self.pe_scan(value) or self.ml_scan(value):
-                return True
+            for entry in psutil.Process(pid).memory_maps():
+                file = entry.path.replace("\\", "/")
+                if ":/Windows" not in file and ":/Program" not in file:
+                    if self.api_scan(file) or self.pe_scan(file):
+                        return True
             return False
         except:
             return False
@@ -1224,48 +1192,41 @@ class MainWindow_Controller(QMainWindow):
             time.sleep(0.05)
             for p in psutil.process_iter():
                 try:
-                    name, file, cmd = p.name(), p.exe().replace("\\", "/"), p.cmdline()
-                    if p.pid not in existing_processes and file not in self.whitelist:
+                    if p.pid not in existing_processes:
                         existing_processes.add(p.pid)
-                        psutil.Process(p.pid).suspend()
-                        if "powershell" in name and self.scan_proc(cmd[-1].split("'")[-2]):
-                            p.kill()
-                            self.send_notify(self.trans("惡意腳本攔截: ")+str(cmd[-1].split("'")[-2]))
-                        elif "cmd.exe" in name and self.scan_proc(" ".join(cmd[2:])):
-                            p.kill()
-                            self.send_notify(self.trans("惡意腳本攔截: ")+str(" ".join(cmd[2:])))
-                        elif ":/Program" not in file and self.scan_proc(cmd[-1]):
-                            p.kill()
-                            self.send_notify(self.trans("惡意軟體攔截: ")+str(cmd[-1]))
-                        elif ":/Windows" not in file and self.scan_proc(p.pid):
-                            p.kill()
-                            self.send_notify(self.trans("惡意軟體攔截: ")+file)
-                        elif file != self.pyas and self.sign_scan(file):
-                            self.proc = p
-                        psutil.Process(p.pid).resume()
-                        gc.collect()
+                        name, file, cmd = p.name(), p.exe().replace("\\", "/"), p.cmdline()
+                        if file != self.pyas and file not in self.whitelist:
+                            psutil.Process(p.pid).suspend()
+                            if "powershell" in name and self.api_scan(cmd[-1].split("'")[-2]):
+                                p.kill()
+                                self.send_notify(self.trans("惡意腳本攔截: ")+str(cmd[-1].split("'")[-2]))
+                            elif "cmd.exe" in name and self.api_scan(" ".join(cmd[2:])):
+                                p.kill()
+                                self.send_notify(self.trans("惡意腳本攔截: ")+str(" ".join(cmd[2:])))
+                            elif ":/Windows" in file and self.api_scan(cmd[-1]):
+                                p.kill()
+                                self.send_notify(self.trans("惡意軟體攔截: ")+str(cmd[-1]))
+                            elif self.proc_scan(p.pid):
+                                p.kill()
+                                self.send_notify(self.trans("惡意軟體攔截: ")+file)
+                            elif self.sign_scan(file):
+                                self.proc = p
+                            psutil.Process(p.pid).resume()
+                            gc.collect()
                 except:
                     pass
 
     def protect_file_thread(self):
         hDir = win32file.CreateFile("C:/Users/",win32con.GENERIC_READ,win32con.FILE_SHARE_READ|win32con.FILE_SHARE_WRITE|win32con.FILE_SHARE_DELETE,None,win32con.OPEN_EXISTING,win32con.FILE_FLAG_BACKUP_SEMANTICS,None)
-        self.ransom_block = False
         while self.file_protect:
             for action, file in win32file.ReadDirectoryChangesW(hDir,1048576,True,win32con.FILE_NOTIFY_CHANGE_FILE_NAME|win32con.FILE_NOTIFY_CHANGE_DIR_NAME|win32con.FILE_NOTIFY_CHANGE_ATTRIBUTES|win32con.FILE_NOTIFY_CHANGE_SIZE|win32con.FILE_NOTIFY_CHANGE_LAST_WRITE|win32con.FILE_NOTIFY_CHANGE_SECURITY,None,None):
                 try:
                     full_path = str(f"C:/Users/{file}").replace("\\", "/")
                     file_type = str(f".{full_path.split('.')[-1]}").lower()
-                    if action == 2 and "/AppData/" not in full_path and file_type in alist:
-                        if self.ransom_block:
-                            self.proc.kill()
-                            self.ransom_block = False
-                            self.send_notify(self.trans("勒索軟體攔截: ")+self.proc.exe().replace("\\", "/"))
-                        self.ransom_block = True
-                    #elif action == 3 and "/AppData/" not in full_path and file_type in slist:
-                        #if os.path.getsize(full_path) <= 20971520:
-                            #if self.sign_scan(full_path) and self.api_scan(full_path):
-                                #self.lock_file(full_path)
-                                #self.send_notify(self.trans("惡意軟體攔截: ")+full_path)
+                    if action == 3 and file_type in slist and os.path.getsize(full_path) <= 20971520:
+                        if os.path.isfile(full_path) and self.api_scan(full_path):
+                            os.remove(full_path)
+                            self.send_notify(self.trans("惡意軟體刪除: ")+full_path)
                 except:
                     pass
 
@@ -1300,9 +1261,9 @@ class MainWindow_Controller(QMainWindow):
         while self.net_protect:
             try:
                 time.sleep(0.5)
-                local_host = socket.gethostbyname(socket.gethostname())
+                local = socket.gethostbyname(socket.gethostname())
                 for conn in self.proc.connections():
-                    if conn.laddr.ip == local_host:
+                    if conn.laddr.ip == local:
                         self.proc.kill()
                         self.send_notify(self.trans("網路通訊攔截: ")+self.proc.exe().replace("\\", "/"))
             except:

PYAS_Version.py

@@ -1,7 +1,7 @@
 import pyinstaller_versionfile
 pyinstaller_versionfile.create_versionfile(
 output_file="versionfile.txt",
-version='2.9.6',
+version='2.9.8',
 company_name="PYAS Security",
 file_description="Python Antivirus Software",
 internal_name="PYAS",

PYML.py

@@ -1,166 +0,0 @@
-from collections import defaultdict
-import math, json, time, types, random
-
-class ListClassifier:
-    def __init__(self):
-        self.word_weights = defaultdict(lambda: {'ones': 0, 'zero': 0})
-        self.total_counts = {'ones': 0, 'zero': 0}
-        self.prior_probs = {'ones': 0, 'zero': 0}
-        self.class_weights = {'ones': 1, 'zero': 1}
-        self.min_quantity = 0
-        self.max_quantity = 1.0
-        self.learning_rate = 0.1
-        self.max_sequence = 100
-        self.test_data = []
-        self.train_data = []
-
-    def load_model(self, model_data):
-        if isinstance(model_data, str):
-            with open(model_data, 'r') as f:
-                model_data = json.load(f)
-        self.word_weights = defaultdict(lambda: {'ones': 0, 'zero': 0})
-        self.word_weights.update(model_data['word_weights'])
-        self.total_counts = model_data['total_counts']
-        self.prior_probs = model_data['prior_probs']
-        self.max_sequence = model_data.get('max_sequence')
-
-    def save_model(self, filename):
-        model_data = {
-            'word_weights': dict(self.word_weights),
-            'total_counts': self.total_counts,
-            'prior_probs': self.prior_probs,
-            'max_sequence': self.max_sequence}
-        with open(filename, 'w') as f:
-            json.dump(model_data, f)
-        print(f"\nModel Is Safe In {filename}")
-
-    def train_model(self, ones_data, zero_data, epochs=1, batch_size=32, min_quantity=0, max_quantity=1.0, max_sequence=10, test_size=0.2, learning_rate=None, random_seed=None, class_weights=None):
-        self.min_quantity = min_quantity
-        self.max_quantity = max_quantity
-        self.max_sequence = max_sequence
-        self.learning_rate = learning_rate if learning_rate is not None else self.learning_rate
-        self.class_weights = class_weights if class_weights is not None else self.class_weights
-        is_generator = isinstance(ones_data[0], types.GeneratorType)
-        all_data = ones_data + zero_data
-        random.seed(random_seed)
-        random.shuffle(all_data)
-        split_index = int((1 - test_size) * len(all_data))
-        self.train_data = all_data[:split_index]
-        self.test_data = all_data[split_index:]
-        train_accuracies, train_losses = [], []
-        test_accuracies, test_losses = [], []
-        for epoch in range(epochs):
-            print(f'Epoch: {epoch + 1}/{epochs}')
-            train_accuracy, train_loss = self._train_epoch(self.train_data, batch_size, is_generator, prefix='Training:')
-            test_accuracy, test_loss = self._test_epoch(self.test_data, batch_size, is_generator, prefix='Testing: ')
-            train_accuracies.append(train_accuracy)
-            train_losses.append(train_loss)
-            test_accuracies.append(test_accuracy)
-            test_losses.append(test_loss)
-        self._print_table(train_accuracies, train_losses, test_accuracies, test_losses)
-        total_lists = len(self.train_data) + len(self.test_data)
-        self.prior_probs['ones'] = len(ones_data) / total_lists
-        self.prior_probs['zero'] = len(zero_data) / total_lists
-        self._filter_word_weights()
-
-    def _train_epoch(self, data, batch_size, is_generator, prefix=''):
-        start_time = time.time()
-        correct_predictions, total_samples, total_loss = 0, 0, 0
-        for i in range(0, len(data), batch_size):
-            batch = data[i:i + batch_size]
-            for lst_gen in batch:
-                lst = list(lst_gen) if is_generator else lst_gen
-                correct_predictions += self._update_weights(lst, 'ones')
-                total_samples += len(lst)
-                total_loss += self._calculate_loss(lst, 'ones')
-            elapsed_time = time.time() - start_time
-            suffix = f'(Accuracy: {correct_predictions/total_samples:.2%}, Loss: {total_loss/total_samples:.4f}, Elapsed: {elapsed_time:.2f}s)'
-            self._print_progress_bar(i + batch_size, len(data), prefix=prefix, suffix=suffix)
-        print('')
-        accuracy = correct_predictions / total_samples
-        average_loss = total_loss / total_samples
-        return accuracy, average_loss
-
-    def _test_epoch(self, data, batch_size, is_generator, prefix=''):
-        start_time = time.time()
-        correct_predictions, total_samples, total_loss = 0, 0, 0
-        for i in range(0, len(data), batch_size):
-            batch = data[i:i + batch_size]
-            for lst_gen in batch:
-                lst = list(lst_gen) if is_generator else lst_gen
-                correct_predictions += (len(lst) - self._update_weights(lst, 'zero'))
-                total_samples += len(lst)
-                total_loss += self._calculate_loss(lst, 'zero')
-            elapsed_time = time.time() - start_time
-            suffix = f'(Accuracy: {correct_predictions/total_samples:.2%}, Loss: {total_loss/total_samples:.4f}, Elapsed: {elapsed_time:.2f}s)'
-            self._print_progress_bar(i + batch_size, len(data), prefix=prefix, suffix=suffix)
-        print('')
-        accuracy = correct_predictions / total_samples
-        average_loss = total_loss / total_samples
-        return accuracy, average_loss
-
-    def _print_table(self, train_accuracies, train_losses, test_accuracies, test_losses):
-        print("\nEpoch\tTrain Accuracy\tTrain Loss\tTest Accuracy\tTest Loss (Table):")
-        for i, (train_acc, train_loss, test_acc, test_loss) in enumerate(zip(train_accuracies, train_losses, test_accuracies, test_losses), 1):
-            print(f"{i}\t{train_acc:.2%}\t\t{train_loss:.4f}\t\t{test_acc:.2%}\t\t{test_loss:.4f}")
-
-    def _print_progress_bar(self, iteration, total, prefix='', suffix='', length=30, fill='█'):
-        percent = min(100.0, max(0.0, 100 * (iteration / float(total))))
-        percent_string = "{0:.2f}".format(percent)
-        filled_length = int(length * iteration // total)
-        bar = fill * filled_length + ' ' * (length - filled_length)
-        print(f'\r    {prefix} |{bar}| {percent_string}% {suffix}', end='\r')
-
-    def _filter_word_weights(self):
-        for word, weights in list(self.word_weights.items()):
-            total_weight = weights['ones'] + weights['zero']
-            if (total_weight < self.min_quantity or total_weight > self.max_quantity * max(self.total_counts.values())):
-                del self.word_weights[word]
-
-    def _get_data_for_word(self, word, category):
-        data = []
-        if isinstance(self.word_weights[word][category], list):
-            for lst in self.word_weights[word][category]:
-                data.append(lst)
-        return data
-
-    def predict(self, new_list):
-        ones_log_prob = self._calculate_category_log_prob(new_list, 'ones')
-        zero_log_prob = self._calculate_category_log_prob(new_list, 'zero')
-        if ones_log_prob > zero_log_prob:
-            return 1
-        else:
-            return 0
-
-    def _update_weights(self, lst, category):
-        correct_prediction = 0
-        for i, word in enumerate(lst):
-            position_encoding = min(i / len(lst), 1.0)
-            word_frequency = self.word_weights[word][category]
-            learning_rate = self.learning_rate / (word_frequency + 1)
-            self.word_weights[word][category] += learning_rate * self.class_weights[category] * position_encoding
-            self.total_counts[category] += learning_rate
-            prediction = self.predict([word])
-            correct_prediction += 1 if prediction == int(category == 'ones') else 0
-        return correct_prediction
-
-    def _calculate_loss(self, lst, category):
-        loss = 0.0
-        for word in lst:
-            if word in self.word_weights:
-                word_weight_category = self.word_weights[word][category]
-                total_weight_category = self.total_counts[category]
-                prob = (word_weight_category + 1) / (total_weight_category + len(self.word_weights))
-                loss += -math.log(prob)
-        return loss
-
-    def _calculate_category_log_prob(self, lst, category):
-        log_prob = 0.0
-        for word in lst:
-            if word in self.word_weights:
-                word_weight_category = self.word_weights[word][category]
-                total_weight_category = self.total_counts[category]
-                prob = (word_weight_category + 1) / (total_weight_category + len(self.word_weights))
-                log_prob += math.log(prob)
-        log_prob += math.log(self.prior_probs[category] + 1e-10)
-        return log_prob