Merge branch 'main' into ruff

.gitignore

@@ -2,6 +2,7 @@ certs/
 savedfiles/
 settings.ini
 .env
+*.crt
 
 ### Python ###
 # Byte-compiled / optimized / DLL files

README.md

@@ -36,6 +36,7 @@ CERT_KEY=cert.key
 CERT_DIR=certs/
 OPENSSL_BINPATH=openssl
 CLIENT_ENCODING=utf-8
+USE_EXTENSIONS=wayback.Wayback,bio.PyBio
 ```
 
 - (Optional) Create a certificate for SSL decryption

base.py

@@ -3,16 +3,17 @@
 # base.py
 # base (common) file
 #
-# Caterpillar Proxy - The simple and parasitic web proxy SPAM spam filter
+# Caterpillar Proxy - The simple web debugging proxy (formerly, php-httpproxy)
 # Namyheon Go (Catswords Research) <gnh1201@gmail.com>
 # https://github.com/gnh1201/caterpillar
 # Created at: 2024-05-20
-# Updated at: 2024-05-21
+# Updated at: 2024-07-09
 #
 
 import hashlib
 import json
 import re
+import importlib
 
 client_encoding = 'utf-8'
 
@@ -71,8 +72,16 @@ class Extension():
         cls.buffer_size = _buffer_size
 
     @classmethod
-    def register(cls, f):
-        cls.extensions.append(f)
+    def register(cls, s):
+        module_name, class_name = s.strip().split('.')[0:2]
+        module_path = 'plugins.' + module_name
+
+        try:
+            module = importlib.import_module(module_path)
+            _class = getattr(module, class_name)
+            cls.extensions.append(_class())
+        except (ImportError, AttributeError) as e:
+            raise ImportError(class_name + " in the extension " + module_name)
 
     @classmethod
     def get_filters(cls):

plugins/bio.py

@@ -0,0 +1,107 @@
+#!/usr/bin/python3
+#
+# bio.py
+# Biopython plugin for Caterpillar Proxy
+#
+# Euiseo Cha (Wonkwang University) <zeroday0619_dev@outlook.com>
+# https://github.com/gnh1201/caterpillar
+# Created at: 2024-07-02
+# Updated at: 2024-07-02
+#
+
+import json
+from Bio.Seq import Seq
+from Bio.SeqUtils import gc_fraction
+
+from base import Extension
+
+def _analyze_sequence(sequence) -> dict[str, str]:
+    """
+    Analyze a given DNA sequence to provide various nucleotide transformations and translations.
+
+    :param sequence: DNA sequence (string) to be analyzed.
+    :return: Dictionary containing the following analyses of the sequence:
+             - complement: DNA complement of the sequence.
+             - complement_rna: RNA complement of the sequence.
+             - reverse_complement: Reverse complement of the DNA sequence.
+             - reverse_complement_rna: Reverse complement of the RNA sequence.
+             - transcription: Transcription of the DNA sequence to RNA.
+             - translation: Translation of the RNA sequence to an amino acid sequence.
+             - back_transcribe: Back-transcription of the RNA sequence to DNA.
+    """
+    sequence_object = Seq(sequence)
+    return dict(
+        complement=str(sequence_object.complement()),
+        complement_rna=str(sequence_object.complement_rna()),
+        reverse_complement=str(sequence_object.reverse_complement()),
+        reverse_complement_rna=str(sequence_object.reverse_complement_rna()),
+        transcription=str(sequence_object.transcribe()),
+        translation=str(sequence_object.translate()),
+        back_transcribe=str(sequence_object.back_transcribe()),
+    )
+
+
+def _gc_content_calculation(sequence) -> dict[str, str]:
+    """
+    Calculate the GC content of a given DNA sequence and return it as a float.
+
+    :param sequence: DNA sequence (string) for which to calculate the GC content.
+    :return: Dictionary containing the GC content as a float.
+    """
+    gc_content = gc_fraction(sequence)
+    return dict(
+        gc_content=gc_content,
+    )
+
+
+class PyBio(Extension):
+    def __init__(self):
+        self.type = "rpcmethod"
+        self.method = "analyze_sequence_init"
+        self.exported_methods = ["analyze_sequence", "gc_content_calculation"]
+
+    def dispatch(self, type, id, params, conn):
+        conn.send(b'Greeting! dispatch')
+
+    def analyze_sequence(self, type, id, params, conn):
+        """
+        Analyze a DNA sequence provided in the params dictionary.
+
+        :param type: Not used in this function.
+        :param id: Not used in this function.
+        :param params: Dictionary containing the DNA sequence with the key "sequence".
+                       Example: {"sequence": "ATGCGTACGTAGCTAGCTAGCGTAGCTAGCTGACT"}
+        :param conn: Not used in this function.
+        :return: Dictionary containing various analyses of the DNA sequence:
+                 - back_transcribe: Back-transcription of the RNA sequence to DNA.
+                 - complement: DNA complement of the sequence.
+                 - complement_rna: RNA complement of the sequence.
+                 - reverse_complement: Reverse complement of the DNA sequence.
+                 - reverse_complement_rna: Reverse complement of the RNA sequence.
+                 - transcription: Transcription of the DNA sequence to RNA.
+                 - translation: Translation of the RNA sequence to an amino acid sequence.
+                 Example: {"back_transcribe": "ATGCGTACGTAGCTAGCTAGCGTAGCTAGCTGACT",
+                           "complement": "TACGCATGCATCGATCGATCGCATCGATCGACTGA",
+                           "complement_rna": "UACGCAUGCAUCGAUCGAUCGCAUCGAUCGACUGA",
+                           "reverse_complement": "AGTCAGCTAGCTACGCTAGCTAGCTACGTACGCAT",
+                           "reverse_complement_rna": "AGUCAGCUAGCUACGCUAGCUAGCUACGUACGCAU",
+                           "transcription": "AUGCGUACGUAGCUAGCUAGCGUAGCUAGCUGACU",
+                           "translation": "MRT*LASVAS*"}
+        """
+        result = _analyze_sequence(params['sequence'])
+        return result
+
+    def gc_content_calculation(self, type, id, params, conn):
+        """
+        Calculate the GC content for a given DNA sequence provided in the params dictionary.
+
+        :param type: Not used in this function.
+        :param id: Not used in this function.
+        :param params: Dictionary containing the DNA sequence with the key "sequence".
+                       Example: {"sequence": "ATGCGTACGTAGCTAGCTAGCGTAGCTAGCTGACT"}
+        :param conn: Not used in this function.
+        :return: Dictionary containing the GC content as a float.
+                 Example: {"gc_content": 0.5142857142857142}
+        """
+        result = _gc_content_calculation(params['sequence'])
+        return result

plugins/container.py

@@ -7,12 +7,12 @@
 # Namyheon Go (Catswords Research) <gnh1201@gmail.com>
 # https://github.com/gnh1201/caterpillar
 # Created at: 2024-03-04
-# Updated at: 2024-03-13
+# Updated at: 2024-07-06
 #
 
 import docker
 
-from server import Extension
+from base import Extension
 
 class Container(Extension):
     def __init__(self):

plugins/fediverse.py

@@ -8,7 +8,7 @@
 # https://github.com/gnh1201/caterpillar
 #
 # Created in: 2022-10-06
-# Updated in: 2024-06-05
+# Updated in: 2024-07-06
 #
 
 import io
@@ -19,7 +19,7 @@ import os.path
 from decouple import config
 from PIL import Image
 
-from server import Extension
+from base import Extension
 
 try:
     client_encoding = config('CLIENT_ENCODING', default='utf-8')

plugins/nmap.py

@@ -0,0 +1,34 @@
+#!/usr/bin/python3
+#
+# portscan.py
+# NMAP port scanning wrapper for Caterpillar Proxy
+#
+# Caterpillar Proxy - The simple web debugging proxy (formerly, php-httpproxy)
+# Namyheon Go (Catswords Research) <gnh1201@gmail.com>
+# https://github.com/gnh1201/caterpillar
+# Created at: 2022-01-26 (github.com/gnh1201/welsonjs)
+# Updated at: 2024-07-09
+#
+import sys
+import nmap
+import json
+
+from base import Extension
+
+class PortScanner(Extension):
+    def __init__(self):
+        self.type = "rpcmethod"
+        self.method = "scan_ports_by_hosts"
+        self.exported_methods = []
+    
+    def dispatch(self, type, id, params, conn):
+        hosts = params['hosts']
+        binpath = params['binpath']
+
+        nm = nmap.PortScanner(nmap_search_path=(binpath,))
+        result = nm.scan(hosts=hosts, arguments='-T5 -sV -p0-65535 --max-retries 0')
+
+        return result;
+
+if __name__ == "__main__":
+    main(sys.argv)

plugins/wayback.py

@@ -7,12 +7,13 @@
 # Namyheon Go (Catswords Research) <gnh1201@gmail.com>
 # https://github.com/gnh1201/caterpillar
 # Created at: 2024-03-13
-# Updated at: 2024-03-13
+# Updated at: 2024-07-06
 #
 
 import requests
+from decouple import config
 
-from server import Extension
+from base import Extension
 
 try:
     client_encoding = config('CLIENT_ENCODING')

server.py

@@ -7,7 +7,7 @@
 # Namyheon Go (Catswords Research) <gnh1201@gmail.com>
 # https://github.com/gnh1201/caterpillar
 # Created at: 2022-10-06
-# Updated at: 2024-06-20
+# Updated at: 2024-07-09
 #
 
 import argparse
@@ -23,7 +23,6 @@ import time
 import hashlib
 import traceback
 import textwrap
-import importlib
 from datetime import datetime
 from platform import python_version
 
@@ -48,6 +47,7 @@ try:
     client_encoding = config('CLIENT_ENCODING', default='utf-8')
     local_domain = config('LOCAL_DOMAIN', default='')
     proxy_pass = config('PROXY_PASS', default='')
+    use_extensions = config('USE_EXTENSIONS', default='')
 except KeyboardInterrupt:
     print("\n[*] User has requested an interrupt")
     print("[*] Application Exiting.....")
@@ -499,9 +499,8 @@ def start():    #Main Program
 
 if __name__== "__main__":
     # load extensions
-    #Extension.register(importlib.import_module("plugins.fediverse").Fediverse())
-    #Extension.register(importlib.import_module("plugins.container").Container())
-    #Extension.register(importlib.import_module("plugins.wayback").Wayback())
+    for s in use_extensions.split(','):
+        Extension.register(s)
 
     # start Caterpillar
     start()

web.py

@@ -7,7 +7,7 @@
 # Namyheon Go (Catswords Research) <gnh1201@gmail.com>
 # https://github.com/gnh1201/caterpillar
 # Created at: 2024-05-20
-# Updated at: 2024-05-20
+# Updated at: 2024-07-10
 #
 
 from flask import Flask, request, redirect, url_for, render_template
@@ -94,6 +94,7 @@ if __name__ == "__main__":
     Extension.set_protocol('http')
 
     # load extensions
-    #Extension.register(importlib.import_module("plugins.yourownplugin").YourOwnPlugin())
+    for s in use_extensions.split(','):
+        Extension.register(s)
 
     app.run(debug=True, host='0.0.0.0', port=listening_port)