Compare commits

..

3 Commits

Author SHA1 Message Date
b32f9b9ed9 bugfix (wow this code is trash 2026-06-18 15:52:21 -04:00
245ba2980d add a hole punch test server 2026-05-24 17:27:25 -04:00
fdef38618b a 2026-05-15 16:31:39 -04:00
4 changed files with 58 additions and 14 deletions

5
.gitignore vendored
View File

@@ -2,4 +2,7 @@ chats.json
config.json
.vscode/
__pycache__/
.venv/
.venv/
build/
dist/
p2pchat.spec

View File

@@ -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

View 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
View File

@@ -15,6 +15,8 @@ import ipaddress
import pyperclip
import requests
HOLE_PUNCH_TEST = ("195.201.227.193", 41234)
class SetupApp(App):
DEFAULT_CSS = """
@@ -446,12 +448,35 @@ class SettingsScreen(Screen):
id="instructions",
)
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):
if event.button.id == "close-btn":
self.dismiss()
elif event.button.id == "add-contact":
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":
valid_ip = True
valid_port = True
@@ -762,23 +787,21 @@ class ChatApp(App):
self.sock.settimeout(None)
self.connected = True
if self.other_name != chat["name"]:
with open("chats.json", "r") as f:
current_chats_config = json.load(f)
# with open("chats.json", "r") as f:
# current_chats_config = json.load(f)
self.call_from_thread(
self.add_message,
"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:
if (
file_config["key"] == local_config["key"]
and file_config["ip"] == local_config["ip"]
):
file_config["name"] = self.other_name
local_config["name"] = self.other_name
break
for local_config in self.chats:
if (
local_config["key"] == chat["key"]
and local_config["ip"] == chat["ip"]
):
local_config["name"] = self.other_name
break
with open("chats.json", "w") as f:
json.dump(current_chats_config, f)
json.dump(self.chats, f)
index = self.chats.index(local_config)
self.call_from_thread(
lambda: list(self.query_one("#contacts").query("ListItem"))[index]