alienhunter/packetserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added a user database and added a few automatic features. Users are created for everyone who connects. User database is publicly searchable for users not marked hidden. Need to add a few more fields for users.

Browse Source
This commit is contained in:
Michael Woods
2025-01-05 01:11:59 -05:00
parent 754a1c580f
commit 7a38f4aa6b
4 changed files with 206 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/packetserver/server/__init__.py
View File

@@ -1,6 +1,7 @@
import pe.app
from packetserver.common import Response, Message, Request, PacketServerConnection, send_response, send_blank_response
from packetserver.server.constants import default_server_config
from packetserver.server.users import User
from copy import deepcopy
import ax25
from pathlib import Path
@@ -49,10 +50,18 @@ class Server:
self.db = ZODB.DB(self.storage)
with self.db.transaction() as conn:
if 'config' not in conn.root():
logging.debug("no config, writing blank default config")
conn.root.config = PersistentMapping(deepcopy(default_server_config))
conn.root.config['blacklist'] = PersistentList()
if 'SYSTEM' not in conn.root.config['blacklist']:
logging.debug("Adding 'SYSTEM' to blacklist in case someone feels like violating FCC rules.")
conn.root.config['blacklist'].append('SYSTEM')
if 'users' not in conn.root():
conn.root.users = OOBTree()
logging.debug("users missing, creating bucket")
conn.root.users = PersistentMapping()
if 'SYSTEM' not in conn.root.users:
logging.debug("Creating system user for first time.")
User('SYSTEM', hidden=True, enabled=False).write_new(conn.root())
init_bulletins(conn.root())
self.app = pe.app.Application()
PacketServerConnection.receive_subscribers.append(lambda x: self.server_receiver(x))
@@ -66,19 +75,29 @@ class Server:
return str(Path(self.home_dir).joinpath('data.zopedb'))
def server_connection_bouncer(self, conn: PacketServerConnection):
logging.debug("new connection bouncer checking for blacklist")
logging.debug("new connection bouncer checking user status")
# blacklist check
blacklisted = False
base = ax25.Address(conn.remote_callsign).call
with self.db.transaction() as storage:
if 'blacklist' in storage.root.config:
bl = storage.root.config['blacklist']
logging.debug(f"A blacklist exists: {bl}")
base = ax25.Address(conn.remote_callsign).call
logging.debug(f"Checking callsign {base.upper()}")
if base.upper() in bl:
logging.debug(f"Connection from blacklisted callsign {base}")
conn.closing = True
blacklisted = True
# user object check
if base in storage.root.users:
logging.debug(f"User {base} exists in db.")
u = storage.root.users[base]
u.seen()
else:
logging.info(f"Creating new user {base}")
u = User(base.upper().strip())
u.write_new(storage.root())
if blacklisted:
count = 0
while count < 10:

23
src/packetserver/server/bulletin.py
View File

@@ -4,8 +4,7 @@ import persistent.list
from persistent.mapping import PersistentMapping
import datetime
from typing import Self,Union,Optional
from packetserver.common import PacketServerConnection, Request, Response, Message
from packetserver.server.requests import send_response, send_blank_response
from packetserver.common import PacketServerConnection, Request, Response, Message, send_response, send_blank_response
import ZODB
import logging
@@ -41,8 +40,8 @@ class Bulletin(persistent.Persistent):
self.author = author
self.subject = subject
self.body = text
self.created_at = None
self.updated_at = None
self.created_at = datetime.datetime.now(datetime.UTC)
self.updated_at = datetime.datetime.now(datetime.UTC)
self.id = None
@classmethod
@@ -78,6 +77,7 @@ class Bulletin(persistent.Persistent):
def handle_bulletin_get(req: Request, conn: PacketServerConnection, db: ZODB.DB):
response = Response.blank()
sp = req.path.split("/")
logging.debug(f"bulletin get path: {sp}")
bid = None
limit = None
if 'limit' in req.vars:
@@ -90,14 +90,18 @@ def handle_bulletin_get(req: Request, conn: PacketServerConnection, db: ZODB.DB)
bid = int(req.vars['id'])
except ValueError:
pass
if len(sp) > 2:
if len(sp) > 1:
logging.debug(f"checking path for bulletin id")
try:
bid = int(sp[2].strip())
logging.debug(f"{sp[1]}")
bid = int(sp[1].strip())
except ValueError:
pass
logging.debug(f"bid is {bid}")
with db.transaction() as db:
if bid:
if bid is not None:
logging.debug(f"retrieving bulletin: {bid}")
bull = Bulletin.get_bulletin_by_id(bid, db.root())
if bull:
response.payload = bull.to_dict()
@@ -105,6 +109,7 @@ def handle_bulletin_get(req: Request, conn: PacketServerConnection, db: ZODB.DB)
else:
response.status_code = 404
else:
logging.debug(f"retrieving all bulletins")
bulls = Bulletin.get_recent_bulletins(db.root(), limit=limit)
response.payload = [bulletin.to_dict() for bulletin in bulls]
response.status_code = 200
@@ -125,13 +130,13 @@ def handle_bulletin_post(req: Request, conn: PacketServerConnection, db: ZODB.DB
b.write_new(db.root())
send_blank_response(conn, req, status_code=201)
def handle_bulletin_update(req: Request, conn: PacketServerConnection, db: ZODB.DB):
def handle_bulletin_update(req: Request, conn: PacketServerConnection, db: ZODB.DB): # TODO
response = Response.blank()
with db.transaction() as db:
pass
send_response(conn, response, req)
def handle_bulletin_delete(req: Request, conn: PacketServerConnection, db: ZODB.DB):
def handle_bulletin_delete(req: Request, conn: PacketServerConnection, db: ZODB.DB): # TODO
response = Response.blank()
with db.transaction() as db:
pass

4
src/packetserver/server/requests.py
View File

@@ -3,6 +3,7 @@
from msgpack.exceptions import OutOfData
from packetserver.common import Message, Request, Response, PacketServerConnection, send_response, send_blank_response
from .bulletin import bulletin_root_handler
from .users import user_root_handler
import logging
from typing import Union
import ZODB
@@ -38,7 +39,8 @@ def root_root_handler(req: Request, conn: PacketServerConnection,
standard_handlers = {
"": root_root_handler,
"bulletin": bulletin_root_handler
"bulletin": bulletin_root_handler,
"user": user_root_handler
}

167
src/packetserver/server/users.py Normal file
View File

@@ -0,0 +1,167 @@
"""Module containing code related to users."""
import ax25
import persistent
import persistent.list
from persistent.mapping import PersistentMapping
import datetime
from typing import Self,Union,Optional
from packetserver.common import PacketServerConnection, Request, Response, Message, send_response, send_blank_response
import ZODB
import logging
import uuid
from uuid import UUID
class User(persistent.Persistent):
def __init__(self, username: str, enabled: bool = True, hidden: bool = False, bio: str = "", status: str = ""):
self._username = username.upper().strip()
self.enabled = enabled
self.hidden = hidden
self.created_at = datetime.datetime.now(datetime.UTC)
self.last_seen = self.created_at
self._uuid = None
if len(bio) > 4000:
self._bio = bio[:4000]
else:
self._bio = bio
if len(status) > 300:
self._status = status[:300]
else:
self._status = status
def write_new(self, db_root: PersistentMapping):
all_uuids = [db_root['users'][x].uuid for x in db_root['users']]
self._uuid = uuid.uuid4()
while self.uuid in all_uuids:
self._uuid = uuid.uuid4()
logging.debug(f"Creating new user account {self.username} - {self.uuid}")
if self.username not in db_root['users']:
db_root['users'][self.username] = self
@property
def uuid(self):
return self._uuid
@classmethod
def get_user_by_username(cls, username: str, db_root: PersistentMapping) -> Self:
try:
if username.upper().strip() in db_root['users']:
return db_root['users'][username.upper().strip()]
except Exception:
return None
return None
@classmethod
def get_user_by_uuid(cls, user_uuid: Union[UUID, bytes, int, str], db_root: PersistentMapping) -> Self:
try:
if type(uuid) is uuid.UUID:
uid = user_uuid
elif type(uuid) is bytes:
uid = uuid.UUID(bytes=user_uuid)
elif type(uuid) is int:
uid = uuid.UUID(int=user_uuid)
else:
uid = uuid.UUID(str(user_uuid))
for user in db_root['users']:
if uid == db_root['users'][user].uuid:
return db_root['users'][user].uuid
except Exception:
return None
return None
@classmethod
def get_all_users(cls, db_root: PersistentMapping, limit: int = None) -> list:
all_users = sorted(db_root['users'].values(), key=lambda user: user.username)
if not limit:
return all_users
else:
if len(all_users) < limit:
return all_users
else:
return all_users[:limit]
def seen(self):
self.last_seen = datetime.datetime.now(datetime.UTC)
@property
def username(self) -> str:
return self._username.upper().strip()
@property
def bio(self) -> str:
return self._bio
@bio.setter
def bio(self, bio: str):
if len(bio) > 4000:
self._bio = bio[:4000]
else:
self._bio = bio
@property
def status(self) -> str:
return self._status
@status.setter
def status(self, status: str):
if len(status) > 300:
self._status = status[:300]
else:
self._status = status
def to_safe_dict(self) -> dict:
return {
"username": self.username,
"status": self.status,
"bio": self.bio,
"last_seen": self.last_seen.isoformat(),
"created_at": self.created_at.isoformat()
}
def handle_user_get(req: Request, conn: PacketServerConnection, db: ZODB.DB):
sp = req.path.split("/")
logging.debug("handle_user_get working")
user = None
user_var = req.vars.get('username')
response = Response.blank()
response.status_code = 404
limit = None
if 'limit' in req.vars:
try:
limit = int(req.vars['limit'])
except ValueError:
pass
with db.transaction() as db:
if len(sp) > 1:
logging.debug(f"trying to get the username from the path {sp[1].strip().upper()}")
user = User.get_user_by_username(sp[1].strip().upper(), db.root())
logging.debug(f"user holds: {user}")
if user and not user.hidden:
response.status_code = 200
response.payload = user.to_safe_dict()
else:
if user_var:
user = User.get_user_by_username(user_var.upper().strip(), db.root())
if user and not user.hidden:
response.status_code = 200
response.payload = user.to_safe_dict()
else:
if user_var:
user = User.get_user_by_username(user_var.upper().strip(), db.root())
if user and not user.hidden:
response.status_code = 200
response.payload = user.to_safe_dict()
else:
response.status_code = 200
response.payload = [x.to_safe_dict() for x in User.get_all_users(db.root(), limit=limit) if not x.hidden]
send_response(conn, response, req)
def handle_user_update(req: Request, conn: PacketServerConnection, db: ZODB.DB): # TODO
pass
def user_root_handler(req: Request, conn: PacketServerConnection, db: ZODB.DB):
logging.debug(f"{req} being processed by user_root_handler")
if req.method is Request.Method.GET:
handle_user_get(req, conn, db)
else:
send_blank_response(conn, req, status_code=404)