150 lines
5.1 KiB
Python
Executable File
150 lines
5.1 KiB
Python
Executable File
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from cryptography.hazmat.backends import default_backend
|
|
import os
|
|
import sys
|
|
import socket
|
|
import datetime
|
|
import struct
|
|
from crccheck.crc import Crc16, CrcArc
|
|
import binascii
|
|
|
|
class StringMaker:
|
|
def __init__(self):
|
|
pass
|
|
|
|
def send_alarm(self, alarm_command, user_id, key, host, port, V):
|
|
key = bytes.fromhex(key)
|
|
now = datetime.datetime.now()
|
|
date_part = now.strftime("%m-%d-%Y")
|
|
time_part = now.strftime("%H:%M:%S")
|
|
input_string = f'"*ADM-CID"0000RA2L0000#{user_id}[#{user_id}|{alarm_command}][V{V}]_{time_part},{date_part}'
|
|
|
|
print(f"Input String: {input_string}")
|
|
|
|
part_before, part_after = input_string.split('[', 1)
|
|
part_before = f"{part_before}["
|
|
part_after = "|" + part_after
|
|
|
|
encryption_part = self.add_padding(part_after)
|
|
print(f"Padded Input: {encryption_part}")
|
|
|
|
encrypted_hex = self.encrypt(encryption_part, key)
|
|
print(f"Encrypted Hex: {encrypted_hex}")
|
|
|
|
before_and_after = f"{part_before}{encrypted_hex}"
|
|
|
|
crc_hash = self.calculate_crc(before_and_after.encode())
|
|
with_hash = f"{crc_hash}{before_and_after}"
|
|
|
|
with_line_feed_and_return = f"\u000A{with_hash}\u000D"
|
|
message_bytes = with_line_feed_and_return.encode()
|
|
|
|
tcp_response = self.send_bytes_with_tcp(host, port, message_bytes)
|
|
return tcp_response.decode() if tcp_response is not None else "No data from Patriot"
|
|
|
|
def to_ascii_hex(self, byte_array):
|
|
return byte_array.hex()
|
|
|
|
|
|
def add_padding(self, input_string):
|
|
# Convert the input string to bytes
|
|
input_bytes = input_string.encode()
|
|
# Calculate the pad length
|
|
pad_length = 16 - len(input_bytes) % 16
|
|
pad_length = pad_length if pad_length != 16 else 0
|
|
# Generate padding with ASCII characters
|
|
padding_ascii = bytes([i % 26 + 65 for i in range(pad_length)])
|
|
# Concatenate padding and input bytes
|
|
padded_bytes = padding_ascii + input_bytes
|
|
|
|
return padded_bytes
|
|
|
|
def get_string_after_first_occurrence(self, input_string, character):
|
|
index = input_string.find(character)
|
|
return input_string[index + 1:] if index != -1 else ""
|
|
|
|
def encrypt(self, input_bytes, key):
|
|
key_bytes = key
|
|
data_bytes = input_bytes
|
|
|
|
# Use the same initialization vector (IV) as in your decryption function
|
|
iv = b'\x00' * 16
|
|
|
|
cipher = Cipher(algorithms.AES(key_bytes), modes.CBC(iv), backend=default_backend())
|
|
encryptor = cipher.encryptor()
|
|
|
|
encrypted_data = encryptor.update(data_bytes) + encryptor.finalize()
|
|
|
|
return binascii.hexlify(encrypted_data).decode('ascii')
|
|
|
|
def decrypt(self, input_bytes, key):
|
|
cipher = Cipher(algorithms.AES(key), modes.CFB(os.urandom(16)), backend=default_backend())
|
|
decryptor = cipher.decryptor()
|
|
decrypted = decryptor.update(input_bytes) + decryptor.finalize()
|
|
return decrypted.decode()
|
|
|
|
def send_bytes_with_tcp(self, ip_address, port, data):
|
|
try:
|
|
with socket.create_connection((ip_address, port), timeout=5) as client:
|
|
client.sendall(data)
|
|
response = client.recv(1024)
|
|
return response
|
|
except Exception as ex:
|
|
print(f"Error: {ex}")
|
|
return None
|
|
|
|
def convert_to_dms_format(self, dd, direction):
|
|
degrees = int(abs(dd))
|
|
minutes = (abs(dd) - degrees) * 60
|
|
seconds = (minutes - int(minutes)) * 60
|
|
sign = 'W' if dd < 0 and direction == "X" else 'S' if dd < 0 and direction == "Y" else 'E' if direction == "X" else 'N'
|
|
return f"{direction}{degrees:03d}{sign}{int(minutes):02d}.{int(round(seconds * 10000000)):07d}"
|
|
|
|
def calculate_crc(self, input_bytes):
|
|
crc_arc = CrcArc()
|
|
crc_arc.process(input_bytes)
|
|
crc_value_hex = crc_arc.finalhex()
|
|
return f"{crc_value_hex.upper()}{len(input_bytes):04X}"
|
|
|
|
arg1 = sys.argv[1] if len(sys.argv) > 1 else None
|
|
arg2 = sys.argv[2] if len(sys.argv) > 2 else None
|
|
arg3 = sys.argv[3] if len(sys.argv) > 3 else None
|
|
host = sys.argv[4] if len(sys.argv) > 4 else None
|
|
arg5 = sys.argv[5] if len(sys.argv) > 5 else None
|
|
signalType = sys.argv[6] if len(sys.argv) > 6 else "PA"
|
|
zone = sys.argv[7] if len(sys.argv) > 7 else "00"
|
|
arg8 = sys.argv[8] if len(sys.argv) > 8 else "0"
|
|
arg9 = sys.argv[9] if len(sys.argv) > 9 else "0"
|
|
arg10 = sys.argv[10] if len(sys.argv) > 10 else "0"
|
|
V = sys.argv[11] if len(sys.argv) > 11 else None
|
|
port = int(arg5)
|
|
|
|
def main():
|
|
global host
|
|
global port
|
|
global arg1
|
|
global arg2
|
|
global arg3
|
|
global signalType
|
|
global zone
|
|
global arg8
|
|
global arg9
|
|
global V
|
|
string_maker = StringMaker()
|
|
key = arg10.encode().hex().upper()
|
|
ClientID = arg1
|
|
SignalType = signalType
|
|
Zone = zone
|
|
if len(Zone) == 1:
|
|
Zone = "00" + Zone
|
|
if len(Zone) == 2:
|
|
Zone = "0" + Zone
|
|
SignalAndZone = SignalType + " 00 " + Zone
|
|
|
|
result = string_maker.send_alarm(SignalAndZone, int(ClientID), key, host, port, V)
|
|
print(result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|