Compare commits
3 Commits
faa9e09af4
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| b32f9b9ed9 | |||
| 245ba2980d | |||
| fdef38618b |
5
.gitignore
vendored
5
.gitignore
vendored
@@ -2,4 +2,7 @@ chats.json
|
|||||||
config.json
|
config.json
|
||||||
.vscode/
|
.vscode/
|
||||||
__pycache__/
|
__pycache__/
|
||||||
.venv/
|
.venv/
|
||||||
|
build/
|
||||||
|
dist/
|
||||||
|
p2pchat.spec
|
||||||
@@ -1 +1,3 @@
|
|||||||
hi
|
if u want to run this then download the repo, have uv python setup and run uv run main.py
|
||||||
|
its a udp hole punching secure messaging app
|
||||||
|
that i dont wanna make a readme for rn
|
||||||
16
extras/hole-punch-test-server.py
Normal file
16
extras/hole-punch-test-server.py
Normal file
@@ -0,0 +1,16 @@
|
|||||||
|
import socket
|
||||||
|
|
||||||
|
SERVER_PORT = 41234
|
||||||
|
SERVER_SEND_PORT = 41235
|
||||||
|
|
||||||
|
sock_recv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||||
|
sock_recv.bind(("0.0.0.0", SERVER_PORT))
|
||||||
|
sock_send = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||||
|
sock_send.bind(("0.0.0.0", SERVER_SEND_PORT))
|
||||||
|
|
||||||
|
print("Running...")
|
||||||
|
|
||||||
|
while True:
|
||||||
|
data, addr = sock_recv.recvfrom(1024)
|
||||||
|
print(f"Recieved test from addr {addr}")
|
||||||
|
sock_send.sendto(data, addr)
|
||||||
47
main.py
47
main.py
@@ -15,6 +15,8 @@ import ipaddress
|
|||||||
import pyperclip
|
import pyperclip
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
|
HOLE_PUNCH_TEST = ("195.201.227.193", 41234)
|
||||||
|
|
||||||
|
|
||||||
class SetupApp(App):
|
class SetupApp(App):
|
||||||
DEFAULT_CSS = """
|
DEFAULT_CSS = """
|
||||||
@@ -446,12 +448,35 @@ class SettingsScreen(Screen):
|
|||||||
id="instructions",
|
id="instructions",
|
||||||
)
|
)
|
||||||
yield Button("Add Contact", id="add-contact")
|
yield Button("Add Contact", id="add-contact")
|
||||||
|
with TabPane("Extras"):
|
||||||
|
yield Label(
|
||||||
|
"This button will run a test to see if this application will work correctly."
|
||||||
|
)
|
||||||
|
yield Button("Test hole punching", id="test-hole")
|
||||||
|
|
||||||
|
def test_hole_punch(self):
|
||||||
|
test_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||||
|
test_sock.bind((my_ip, 0))
|
||||||
|
test_sock.settimeout(5)
|
||||||
|
|
||||||
|
test_sock.sendto(b"TEST", HOLE_PUNCH_TEST)
|
||||||
|
|
||||||
|
try:
|
||||||
|
data, addr = test_sock.recvfrom(1024)
|
||||||
|
except socket.timeout:
|
||||||
|
self.app.call_from_thread(self.notify, "Test failed", severity="error")
|
||||||
|
else:
|
||||||
|
self.app.call_from_thread(self.notify, "Test worked!")
|
||||||
|
test_sock.close()
|
||||||
|
|
||||||
def on_button_pressed(self, event):
|
def on_button_pressed(self, event):
|
||||||
if event.button.id == "close-btn":
|
if event.button.id == "close-btn":
|
||||||
self.dismiss()
|
self.dismiss()
|
||||||
elif event.button.id == "add-contact":
|
elif event.button.id == "add-contact":
|
||||||
self.app.push_screen(ChoiceScreen(), self.on_screen_done)
|
self.app.push_screen(ChoiceScreen(), self.on_screen_done)
|
||||||
|
elif event.button.id == "test-hole":
|
||||||
|
self.notify("Test started....", severity="information")
|
||||||
|
threading.Thread(target=self.test_hole_punch).start()
|
||||||
elif event.button.id == "profile-confirm":
|
elif event.button.id == "profile-confirm":
|
||||||
valid_ip = True
|
valid_ip = True
|
||||||
valid_port = True
|
valid_port = True
|
||||||
@@ -762,23 +787,21 @@ class ChatApp(App):
|
|||||||
self.sock.settimeout(None)
|
self.sock.settimeout(None)
|
||||||
self.connected = True
|
self.connected = True
|
||||||
if self.other_name != chat["name"]:
|
if self.other_name != chat["name"]:
|
||||||
with open("chats.json", "r") as f:
|
# with open("chats.json", "r") as f:
|
||||||
current_chats_config = json.load(f)
|
# current_chats_config = json.load(f)
|
||||||
self.call_from_thread(
|
self.call_from_thread(
|
||||||
self.add_message,
|
self.add_message,
|
||||||
"system: other user has new name, updating... (only you can see this)",
|
"system: other user has new name, updating... (only you can see this)",
|
||||||
)
|
)
|
||||||
for file_config in current_chats_config:
|
for local_config in self.chats:
|
||||||
for local_config in self.chats:
|
if (
|
||||||
if (
|
local_config["key"] == chat["key"]
|
||||||
file_config["key"] == local_config["key"]
|
and local_config["ip"] == chat["ip"]
|
||||||
and file_config["ip"] == local_config["ip"]
|
):
|
||||||
):
|
local_config["name"] = self.other_name
|
||||||
file_config["name"] = self.other_name
|
break
|
||||||
local_config["name"] = self.other_name
|
|
||||||
break
|
|
||||||
with open("chats.json", "w") as f:
|
with open("chats.json", "w") as f:
|
||||||
json.dump(current_chats_config, f)
|
json.dump(self.chats, f)
|
||||||
index = self.chats.index(local_config)
|
index = self.chats.index(local_config)
|
||||||
self.call_from_thread(
|
self.call_from_thread(
|
||||||
lambda: list(self.query_one("#contacts").query("ListItem"))[index]
|
lambda: list(self.query_one("#contacts").query("ListItem"))[index]
|
||||||
|
|||||||
Reference in New Issue
Block a user