before fix pytest

.vscode/settings.json

@@ -10,6 +10,9 @@
     "MD013": false // 長い行
   },
   // python
+  // NOTE: venvを使っている場合は，pythonのinterpreterPathが通っていないとformatterが動かない。
+  "python.defaultInterpreterPath": "/app/.venv/bin/python",
+
   "flake8.args": ["--ignore=E111, E114, E402, E501"],
   "[python]": {
     "editor.defaultFormatter": "ms-python.black-formatter",

README.md

@@ -33,20 +33,20 @@ PORT       STATE SERVICE
 
 ```shell
 docker run my_portscanner -h
-usage: my_portscanner [-h] [-sT CONNECT_SCAN] [-oN FILE_TXT] [-p PORT] [-v]
+usage: my_portscanner [-h] [-sT] [-sS] [-oN FILE_TXT] [-p PORT] [--version]
                       target_ip
 
 positional arguments:
   target_ip             set target ip address.
 
 options:
   -h, --help            show this help message and exit
-  -sT CONNECT_SCAN, --connect_scan CONNECT_SCAN
-                        TCP connect Scan
+  -sT, --connect_scan   TCP connect scan
+  -sS, --stealth_scan   TCP SYN scan
   -oN FILE_TXT, --file_txt FILE_TXT
                         output txt file name.
   -p PORT, --port PORT  port number lists
-  -v, --version         show program's version number and exit
+  --version             show program's version number and exit
 ```
 
 ---

src/my_portscanner/__init__.py

@@ -1,4 +1,4 @@
-# coding: utf-8
+# coding utf-8
 import socket
 import re
 from .options import parse_args
@@ -16,16 +16,16 @@ def main():
     )
 
     # localhostを指定したスキャンはコンテナ内のlocalhostを指すので無効にする。
-    if args["target_ip"] is "localhost" or args["target_ip"] is "127.0.0.1":
-        print("When executed via `docker run`, a scan targeting `localhost` is not valid.")
-        exit()
+    if ((args["target_ip"] == "localhost") or (args["target_ip"] == "127.0.0.1")):
+        print("[WARNING]: When executed via `docker run`, scan targeting `localhost` is not valid.")
 
     # target_ipがipアドレスの形式でない場合はhost名を名前解決する。
-    match = re.match(
-        r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$", args["target_ip"]
-    )
+    match = re.match(r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$", args["target_ip"])
     if not match:
-        target_ip = socket.gethostbyname(args["target_ip"])
+        try:
+            target_ip = socket.gethostbyname(args["target_ip"])
+        except socket.gaierror:
+            print(f'Failed to resolve "{args["target_ip"]}".')
     else:
         target_ip = args["target_ip"]
     print(f"my_portscanner scan report for {args["target_ip"]} ({target_ip})")
@@ -44,4 +44,5 @@ def main():
         print("invalid scan type")
         exit()
 
+
 __all__ = ["main"]

src/my_portscanner/options.py

@@ -13,14 +13,20 @@ def parse_args() -> dict:
     parser = argparse.ArgumentParser()
     parser.add_argument("target_ip", help="set target ip address.", type=str)
     parser.add_argument(
-        "-sT", "--connect_scan", action="store_true", help="TCP connect scan",
+        "-sT",
+        "--connect_scan",
+        action="store_true",
+        help="TCP connect scan",
     )
     parser.add_argument(
-        "-sS", "--stealth_scan", action="store_true", help="TCP SYN scan",
+        "-sS",
+        "--stealth_scan",
+        action="store_true",
+        help="TCP SYN scan",
     )
     parser.add_argument("-oN", "--file_txt", help="output txt file name.", type=str)
     parser.add_argument("-p", "--port", help="port number lists", type=str)
-    parser.add_argument("-v", "--version", action="version", version=__version__)
+    parser.add_argument("--version", action="version", version=__version__)
     p = parser.parse_args()
 
     if p.port is not None:

src/my_portscanner/scan_tools/ConnectScan.py

@@ -11,7 +11,7 @@ def run(self) -> list[int]:
         """_summary_
         run connect scan
 
-        # NOTE: https://docs.python.org/ja/3/library/socket.html#socket.socket.connect_ex によるとconnece_exは例外を送出せずにエラーコードを戻り値として返すため，例外処理をしない。
+        NOTE: https://docs.python.org/ja/3/library/socket.html#socket.socket.connect_ex によるとconnece_exは例外を送出せずにエラーコードを戻り値として返すため，例外処理をしない。
         Returns:
             list[int]: open ports list
         """

src/my_portscanner/scan_tools/Scan.py

@@ -16,10 +16,12 @@ def __init__(self, *args, **kwargs):
             TypeError: _description_
         """
         if type(self) is Scan:
-            raise TypeError("Scan is an abstract class and cannot be instantiated directly")
+            raise TypeError(
+                "Scan is an abstract class and cannot be instantiated directly"
+            )
 
     @abstractmethod
-    def run(self) -> None:
+    def run(self) -> list[int]:
         """_summary_
         port scanを実行するabstract method
         """

src/my_portscanner/scan_tools/SynScan.py

@@ -6,25 +6,32 @@
 
 @dataclass
 class SynScan(Scan):
-
     def run(self) -> list[int]:
-        """_summary_
+        """
         run SYN scan
 
         Returns:
             list[int]: open ports list
         """
         self.open_port_list = []
 
-        # conf.verb = 0 # packet送信時のログをSTDOUTに表示しない
+        conf.verb = 0  # packet送信時のログをSTDOUTに表示しない
         for port in self.target_port_list:
-
             # SYN packetを作成して送信する
-            syn_packet = IP(dst=self.target_ip)/TCP(dport=port, flags="S")
-            response_packet = sr1(syn_packet)
-
+            syn_packet = IP(dst=self.target_ip) / TCP(dport=port, flags="S")
+            try:
+                response_packet = sr1(syn_packet, timeout=1)
+            except PermissionError:
+                print(
+                    "You requested a scan type which requires root privileges.\nQUITTING!"
+                )
+
+            try:
+                response_packet.haslayer(TCP)
+            except AttributeError:
+                print(f"setup_target: failed to determine route to {self.target_ip}")
             # SYN/ACKパケットが返ってきた際にopen portとしてリストに追加
-            if (response_packet.haslayer(TCP) and response_packet[TCP].flags == "SA"):
+            if response_packet[TCP].flags == "SA":
                 self.open_port_list.append(port)
 
         return self.open_port_list

src/tests/scan_tools/test_ConnectScan.py

@@ -6,7 +6,6 @@
 
 
 class TestConnectScan(unittest.TestCase):
-
     def setUp(self):
         """
         テスト実行時に毎回実行され，

src/tests/scan_tools/test_SynScan.py

@@ -4,6 +4,7 @@
 from scapy.all import IP, TCP
 from my_portscanner.scan_tools.SynScan import SynScan
 
+
 class TestSynScan(unittest.TestCase):
     def setUp(self):
         """
@@ -15,7 +16,7 @@ def setUp(self):
         self.target_port_list = [22, 80, 443]
         self.expected_open_ports = [22, 443]
 
-        def sr1_side_effect(packet):
+        def sr1_side_effect(packet, timeout):
             if packet[TCP].dport in self.expected_open_ports:
                 mock_response = MagicMock()
                 mock_response.haslayer.return_value = True
@@ -25,10 +26,12 @@ def sr1_side_effect(packet):
                 mock_response = MagicMock()
                 mock_response.haslayer.return_value = False
                 return mock_response
-        self.sr1_side_effect = sr1_side_effect # test_runから参照するためにインスタンス変数に格納
 
+        self.sr1_side_effect = (
+            sr1_side_effect  # test_runから参照するためにインスタンス変数に格納
+        )
 
-    @patch('my_portscanner.scan_tools.SynScan.sr1')
+    @patch("my_portscanner.scan_tools.SynScan.sr1")
     def test_run(self, mock_sr1):
         scan = SynScan(target_ip=self.target_ip, target_port_list=self.target_port_list)
 
@@ -39,5 +42,5 @@ def test_run(self, mock_sr1):
         self.assertEqual(open_ports, self.expected_open_ports)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     unittest.main()