Projects STRLCPY RTSPbrute Commits 162d252d
🤬
  • ■ ■ ■ ■ ■ ■
    rtspbrute/modules/rtsp.py
    skipped 58 lines
    59 59   raise e
    60 60   
    61 61   if port not in range(65536):
    62  - raise ValueError(f"({port}) is not a valid port")
     62 + raise ValueError(f"{port} is not a valid port")
    63 63   
    64 64   self.ip = ip
    65 65   self.port = port
    skipped 2 lines
    68 68   self.status: Status = Status.NONE
    69 69   self.auth_method: AuthMethod = AuthMethod.NONE
    70 70   self.last_error: Union[Exception, None] = None
    71  - self.realm: Union[str, None] = None
    72  - self.nonce: Union[str, None] = None
     71 + self.realm: str = ""
     72 + self.nonce: str = ""
    73 73   self.socket = None
    74 74   self.timeout = timeout
    75  - self.packet = None
     75 + self.packet = ""
    76 76   self.cseq = 0
    77  - self.data = None
     77 + self.data = ""
    78 78   
    79 79   @property
    80 80   def route(self):
    skipped 17 lines
    98 98   if port is None:
    99 99   port = self.port
    100 100   
    101  - self.packet = None
     101 + self.packet = ""
    102 102   self.cseq = 0
    103  - self.data = None
     103 + self.data = ""
    104 104   retry = 0
    105 105   while retry < MAX_RETRIES and not self.is_connected:
    106 106   try:
    skipped 71 lines
  • ■ ■ ■ ■
    rtspbrute/modules/utils.py
    skipped 83 lines
    84 84   if match:
    85 85   return match.group(1)
    86 86   else:
    87  - return None
     87 + return ""
    88 88   
    89 89   
    90 90  def load_txt(path: Path, name: str) -> List[str]:
    skipped 54 lines
  • ■ ■ ■ ■ ■ ■
    tests/test_rtsp.py
     1 +import socket
     2 +from typing import Type
     3 + 
     4 +import pytest
     5 + 
     6 +from rtspbrute.modules.rtsp import AuthMethod, RTSPClient, Status
     7 + 
     8 + 
     9 +def test_init():
     10 + rtsp = RTSPClient("0.0.0.0")
     11 + assert rtsp.ip == "0.0.0.0"
     12 + assert rtsp.port == 554
     13 + assert rtsp.timeout == 2
     14 + assert rtsp.credentials == ":"
     15 + 
     16 + 
     17 +def test_init_error():
     18 + with pytest.raises(ValueError) as excinfo:
     19 + rtsp = RTSPClient("")
     20 + assert "'' does not appear to be an IPv4 or IPv6 address" in str(excinfo.value)
     21 + 
     22 + with pytest.raises(ValueError) as excinfo:
     23 + rtsp = RTSPClient("0.0.0.0", 65536)
     24 + assert "65536 is not a valid port" in str(excinfo.value)
     25 + 
     26 + 
     27 +def test_route():
     28 + rtsp = RTSPClient("0.0.0.0")
     29 + assert rtsp.route == ""
     30 + rtsp.routes.append("/1")
     31 + rtsp.routes.append("/2")
     32 + assert rtsp.route == "/1"
     33 + 
     34 + 
     35 +def test_is_connected():
     36 + rtsp = RTSPClient("0.0.0.0")
     37 + assert not rtsp.is_connected
     38 + rtsp.status = Status.CONNECTED
     39 + assert rtsp.is_connected
     40 + 
     41 + 
     42 +def test_is_authorized():
     43 + rtsp = RTSPClient("0.0.0.0")
     44 + assert not rtsp.is_authorized
     45 + rtsp.data = "200 OK"
     46 + assert rtsp.is_authorized
     47 + 
     48 + 
     49 +def test_connect_successful(monkeypatch):
     50 + def _create_connection(address, timeout):
     51 + return (address, timeout)
     52 + 
     53 + monkeypatch.setattr(socket, "create_connection", _create_connection)
     54 + 
     55 + rtsp = RTSPClient("0.0.0.0")
     56 + connected = rtsp.connect()
     57 + assert connected
     58 + assert rtsp.socket == (("0.0.0.0", 554), 2)
     59 + assert rtsp.status is Status.CONNECTED
     60 + assert not rtsp.last_error
     61 + 
     62 + 
     63 +def test_connect_unsuccessful(monkeypatch):
     64 + def _create_connection(address, timeout):
     65 + raise TimeoutError(address, timeout)
     66 + 
     67 + monkeypatch.setattr(socket, "create_connection", _create_connection)
     68 + 
     69 + rtsp = RTSPClient("0.0.0.0")
     70 + connected = rtsp.connect(8554)
     71 + assert not connected
     72 + assert not rtsp.socket
     73 + assert rtsp.status is Status.TIMEOUT
     74 + assert rtsp.last_error.args == (("0.0.0.0", 8554), 2)
     75 + 
     76 + 
     77 +def test_authorize_without_connection():
     78 + rtsp = RTSPClient("0.0.0.0")
     79 + authorized = rtsp.authorize()
     80 + assert not authorized
     81 + 
     82 + 
     83 +class MockSocket:
     84 + def __init__(
     85 + self,
     86 + successful: bool = True,
     87 + exception: Type[Exception] = TimeoutError,
     88 + data: str = "",
     89 + ) -> None:
     90 + self.successful = successful
     91 + self.exception = exception
     92 + self.data_to_recv = data
     93 + 
     94 + self.sent_data: bytes
     95 + self.recv_bufsize: int
     96 + 
     97 + def sendall(self, data):
     98 + if self.successful:
     99 + self.sent_data = data
     100 + return
     101 + else:
     102 + raise self.exception()
     103 + 
     104 + def recv(self, bufsize):
     105 + self.recv_bufsize = bufsize
     106 + return self.data_to_recv.encode()
     107 + 
     108 + def close(self):
     109 + return
     110 + 
     111 + 
     112 +def test_authorize_successful():
     113 + rtsp = RTSPClient("0.0.0.0")
     114 + rtsp.status = Status.CONNECTED
     115 + rtsp.socket = MockSocket(
     116 + data="RTSP/1.0 200 OK\r\nCSeq: 1\r\nServer: Hipcam RealServer/V1.0\r\n\r\n"
     117 + )
     118 + authorized = rtsp.authorize()
     119 + assert authorized
     120 + assert rtsp.cseq == 1
     121 + assert rtsp.packet
     122 + assert rtsp.socket.sent_data
     123 + assert rtsp.data
     124 + assert rtsp.auth_method is AuthMethod.NONE
     125 + 
     126 + 
     127 +def test_authorize_unsuccessful():
     128 + rtsp = RTSPClient("0.0.0.0")
     129 + rtsp.status = Status.CONNECTED
     130 + rtsp.socket = MockSocket(successful=False)
     131 + authorized = rtsp.authorize()
     132 + assert not authorized
     133 + assert rtsp.cseq == 1
     134 + assert rtsp.packet
     135 + assert not hasattr(rtsp.socket, "sent_data")
     136 + assert not rtsp.data
     137 + assert rtsp.status is Status.TIMEOUT
     138 + assert rtsp.last_error
     139 + 
     140 + 
     141 +def test_str():
     142 + rtsp = RTSPClient("0.0.0.0")
     143 + assert str(rtsp) == f"rtsp://{rtsp.ip}:{rtsp.port}"
     144 + rtsp.port = 8554
     145 + rtsp.credentials = "admin:admin"
     146 + rtsp.routes.append("/1")
     147 + assert str(rtsp) == f"rtsp://{rtsp.credentials}@{rtsp.ip}:{rtsp.port}{rtsp.route}"
     148 + 
  • ■ ■ ■ ■ ■ ■
    tests/test_utils.py
    skipped 42 lines
    43 43   def test_realm(self):
    44 44   var = "realm"
    45 45   assert utils.find(var, self.response_with_auth) == "Hipcam RealServer/V1.0"
    46  - assert utils.find(var, self.response_without_auth) == None
     46 + assert utils.find(var, self.response_without_auth) == ""
    47 47   
    48 48   def test_nonce(self):
    49 49   var = "nonce"
    50 50   assert utils.find(var, self.response_with_auth) == "somenonce"
    51  - assert utils.find(var, self.response_without_auth) == None
     51 + assert utils.find(var, self.response_without_auth) == ""
    52 52   
    53 53   
    54 54  class TestLoadTxt:
    skipped 77 lines
Please wait...
Page is in error, reload to recover