Compare commits
3 Commits
faa9e09af4
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| b32f9b9ed9 | |||
| 245ba2980d | |||
| fdef38618b |
3
.gitignore
vendored
3
.gitignore
vendored
@@ -3,3 +3,6 @@ config.json
|
||||
.vscode/
|
||||
__pycache__/
|
||||
.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 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]
|
||||
|
||||
Reference in New Issue
Block a user