engineering/upload.unturf.com
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

A test_agent.sh & a README.rst

Browse source 
deleted:    .app.py.swp
	modified:   .gitignore
	new file:   README.rst
	modified:   app.py
	new file:   initialize_db.py
	new file:   test_agent.sh
This commit is contained in:
Russell Ballestrini 2025-01-06 19:48:23 -05:00
parent af2b5acc21
commit 596dbb923d
6 changed files with 547 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

177
app.py
View file

@ -1,3 +1,6 @@
###############################################################################
# app.py - Full pyrafiles Application with Unicode-safe Content-Disposition
###############################################################################
import os
import base64
import datetime
@ -11,6 +14,7 @@ import smtplib
import mimetypes
import json
import logging
import unicodedata
from email.mime.text import MIMEText
from pyramid.config import Configurator
@ -47,6 +51,31 @@ from zope.sqlalchemy import register # Import zope.sqlalchemy
log = logging.getLogger(__name__)
################################################################################
# Environment Variables and Defaults
################################################################################
# If PYRAFILES_SECRET is missing, generate a random 64-character secret.
pyrafiles_secret = os.environ.get("PYRAFILES_SECRET", "")
if not pyrafiles_secret:
pyrafiles_secret = "".join(
random.choices(string.ascii_letters + string.digits, k=64)
)
log.info(f"Generated random PYRAFILES_SECRET: {pyrafiles_secret}")
# Database URL can be overridden by environment variable
APP_DIR = os.path.dirname(os.path.abspath(__file__))
default_main_db_url = f"sqlite:///{os.path.join(APP_DIR, 'main.db')}"
DB_URL = os.environ.get("PYRAFILES_DB_URL", default_main_db_url)
# Host and port for the application
HOST = os.environ.get("PYRAFILES_HOST", "0.0.0.0")
PORT = int(os.environ.get("PYRAFILES_PORT", "6544"))
# SMTP host/port
smtp_host = os.environ.get("PYRAFILES_SMTP_HOST", "localhost")
smtp_port = int(os.environ.get("PYRAFILES_SMTP_PORT", "25"))
################################################################################
# Helper Functions
################################################################################
@ -66,7 +95,6 @@ def get_gravatar_url(email, size=100):
def send_email(to_email, subject, body):
# For testing purposes, print the email content to the console
log.debug("======= Email Sent =======")
log.debug(f"To: {to_email}")
log.debug(f"Subject: {subject}")
@ -79,7 +107,7 @@ def send_email(to_email, subject, body):
msg["To"] = to_email
try:
s = smtplib.SMTP("localhost", 25)
s = smtplib.SMTP(smtp_host, smtp_port)
s.sendmail("noreply@example.com", [to_email], msg.as_string())
s.quit()
except Exception as e:
@ -121,12 +149,12 @@ def get_current_user(request):
# Generate UUID and short ID
user_uuid = uuid.uuid4()
user_id = str(user_uuid)
new_user_id = str(user_uuid)
short_id = uuid_to_short_id(user_uuid)
# Create a new guest user (do not create user database)
# Create a new guest user
guest_user = User(
id=user_id,
id=new_user_id,
short_id=short_id,
email=None, # Guests don't have an email
username=guest_username,
@ -156,17 +184,15 @@ def uuid_to_short_id(u):
def short_id_to_uuid(sid):
"""Decode the short ID back to UUID, trying different padding lengths."""
# Try with 0 to 3 padding characters
for padding_length in range(5):
for padding_length in range(6):
try:
padded = sid + ("=" * padding_length)
bytes_data = base64.urlsafe_b64decode(padded)
if len(bytes_data) == 16: # UUID is 16 bytes
return uuid.UUID(bytes=bytes_data)
except Exception as e:
except Exception:
continue
# If we get here, none of the padding attempts worked
log.error(
f"Failed to convert short_id {sid} to UUID after trying all padding lengths"
)
@ -174,8 +200,7 @@ def short_id_to_uuid(sid):
def get_user_db_url(user_id):
"""Return the database URL for the user's SQLite database."""
APP_DIR = os.path.dirname(os.path.abspath(__file__))
"""Return the database URL for the user's SQLite database, in the same directory."""
db_file = os.path.join(APP_DIR, f"user_{user_id}.db")
return f"sqlite:///{db_file}"
@ -189,12 +214,24 @@ def filesizeformat(value):
return f"{value:.2f} PB"
def sanitize_filename_for_http_header(filename):
"""
Ensure that the filename is safe for Waitress (Latin-1 headers).
Converts to ASCII, replacing or removing characters that won't encode.
"""
normalized = unicodedata.normalize("NFKD", filename)
ascii_bytes = normalized.encode("ascii", "ignore") # drop non-ASCII
safe = ascii_bytes.decode("ascii")
# Replace any remaining bad chars with underscores
# e.g. keep alphanumerics, dots, underscores, hyphens, etc.
safe = re.sub(r"[^A-Za-z0-9._-]+", "_", safe)
return safe or "download"
################################################################################
# Database Setup
################################################################################
APP_DIR = os.path.dirname(os.path.abspath(__file__))
DB_URL = f"sqlite:///{os.path.join(APP_DIR, 'main.db')}"
log.debug(f"Using database URL: {DB_URL}") # For debugging
Base = declarative_base()
@ -212,7 +249,6 @@ class User(Base):
enable_gravatar = Column(Boolean, default=False) # Gravatar support
is_admin = Column(Boolean, default=False) # Admin flag
# Indexes for faster lookup
__table_args__ = (
Index("ix_users_id", "id"),
Index("ix_users_short_id", "short_id"),
@ -235,7 +271,6 @@ class Media(Base):
is_public = Column(Boolean, default=True)
size = Column(Integer, nullable=False) # Size in bytes
# Indexes for faster lookup
__table_args__ = (
Index("ix_media_id", "id"),
Index("ix_media_short_id", "short_id"),
@ -281,7 +316,7 @@ def get_user_dbsession_by_user_id(user_id, request):
register(user_dbsession) # Register with zope.sqlalchemy
# Attach cleanup callbacks
def cleanup(request):
def cleanup(_request):
user_dbsession.remove()
user_engine.dispose()
@ -334,7 +369,7 @@ def login_post_view(request):
is_verified=False,
)
session.add(user)
session.flush() # Use flush instead of commit
session.flush()
# Generate 6-digit code
code_str = f"{random.randint(0,999999):06d}"
@ -407,7 +442,7 @@ def verify_post_view(request):
user_db_url, connect_args={"check_same_thread": False}, poolclass=StaticPool
)
Base.metadata.create_all(user_engine) # Create tables in user's database
user_engine.dispose() # Dispose the engine
user_engine.dispose()
return HTTPFound(location=request.route_url("home"))
@ -432,7 +467,7 @@ def profile_get_view(request):
# Initialize stats
total_uploads = 0
total_size = 0 # in bytes
total_size = 0
user_dbsession = request.user_dbsession
if user_dbsession:
@ -484,10 +519,14 @@ def download_database_view(request):
return Response("Database file not found.", status=404)
with open(db_file, "rb") as f:
data = f.read()
# Safely build a filename
download_filename = sanitize_filename_for_http_header(f"user_{user.id}.db")
response = Response(body=data, content_type="application/octet-stream")
response.headers["Content-Disposition"] = (
f'attachment; filename="user_{user.id}.db"'
)
response.headers[
"Content-Disposition"
] = f'attachment; filename="{download_filename}"'
return response
@ -515,8 +554,8 @@ def export_user_record_view(request):
# Convert to JSON string
user_json = json.dumps(user_record).encode("utf-8")
# Create response
response = Response(body=user_json, content_type="application/json")
# Plain ASCII filename is safe
response.headers["Content-Disposition"] = 'attachment; filename="user_record.json"'
return response
@ -534,9 +573,6 @@ def import_user_record_get_view(request):
@view_config(route_name="import_user_record", request_method="POST")
@admin_required
def import_user_record_post_view(request):
# Admin-only import process
# Get the uploaded user record file
user_record_file = request.POST.get("user_record_file")
if (
user_record_file is None
@ -609,7 +645,7 @@ def upload_media_post_view(request):
if len(raw_bytes) > max_size:
return Response("File size exceeds the 30MB limit.", status=400)
file_size = len(raw_bytes) # Store the size in bytes
file_size = len(raw_bytes)
# Determine media type based on MIME type
filename = media_file.filename
@ -622,6 +658,10 @@ def upload_media_post_view(request):
# Get title from form
title = request.POST.get("title", "").strip()
# If user did not enter a title, default to the original filename
if not title:
title = filename
# Encode content to base64
encoded_str = base64.b64encode(raw_bytes).decode("utf-8")
@ -637,7 +677,7 @@ def upload_media_post_view(request):
short_id=media_short_id,
user_id=user.id,
filename=filename,
title=title,
title=title, # now has a fallback of filename
media_type=media_type,
media_b64=encoded_str,
is_public=is_public,
@ -645,7 +685,6 @@ def upload_media_post_view(request):
)
user_dbsession.add(media)
user_dbsession.flush()
# No need to commit; transaction manager will handle it
return HTTPFound(
location=request.route_url(
@ -658,21 +697,19 @@ def upload_media_post_view(request):
@view_config(route_name="list_media", renderer="list_media.html.j2")
def list_media_view(request):
# Aggregate public media from all users
# Aggregate public media from all verified users
s = request.dbsession
users = s.query(User).filter(User.is_verified == True).all()
media_list = []
for user in users:
# Get user_dbsession
user_dbsession = get_user_dbsession_by_user_id(user.id, request)
if not user_dbsession:
continue # Skip users without database
continue
media_items = user_dbsession.query(Media).filter(Media.is_public == True).all()
for media in media_items:
for m in media_items:
media_list.append(
{
"media": media,
"media": m,
"username": user.username,
"user_short_id": user.short_id,
}
@ -689,12 +726,10 @@ def list_media_view(request):
@view_config(route_name="user_media", renderer="user_media.html.j2")
def user_media_view(request):
# View uploads by a particular user
user_short_id = request.matchdict.get("user_short_id")
log.debug(f"Looking up user with short_id: {user_short_id}") # Debug
log.debug(f"Looking up user with short_id: {user_short_id}")
try:
# First convert short_id to UUID
user_uuid = short_id_to_uuid(user_short_id)
if not user_uuid:
log.error(f"Could not convert short_id {user_short_id} to UUID")
@ -702,7 +737,7 @@ def user_media_view(request):
s = request.dbsession
user = s.query(User).filter_by(id=str(user_uuid)).first()
log.debug(f"User found: {user}") # Debug
log.debug(f"User found: {user}")
if not user:
return Response("User not found.", status=404)
@ -710,22 +745,18 @@ def user_media_view(request):
viewer = request.user
is_owner = viewer and viewer.id == user.id
# Get user_dbsession
user_dbsession = get_user_dbsession_by_user_id(user.id, request)
if not user_dbsession:
return Response("User has no uploads.", status=404)
if is_owner:
# Show all media (public and private)
media_items = user_dbsession.query(Media).all()
else:
# Show only public media
media_items = (
user_dbsession.query(Media).filter(Media.is_public == True).all()
)
# Sort media by upload date (recent first)
media_items.sort(key=lambda media: media.upload_date, reverse=True)
media_items.sort(key=lambda m: m.upload_date, reverse=True)
return {
"request": request,
@ -735,7 +766,7 @@ def user_media_view(request):
}
except Exception as e:
log.exception(f"Error processing user_short_id {user_short_id}")
log.exception(f"Error processing user_short_id {user_short_id}: {e}")
return Response("Error processing request.", status=500)
@ -752,12 +783,10 @@ def view_media_details_view(request):
return Response("User not found.", status=404)
user_id = user.id
# Get user_dbsession
user_dbsession = get_user_dbsession_by_user_id(user_id, request)
if not user_dbsession:
return Response("User database not found.", status=404)
# Lookup media by short_id
media = user_dbsession.query(Media).filter_by(short_id=media_short_id).first()
if not media:
return Response("Media not found.", status=404)
@ -789,16 +818,13 @@ def delete_media_view(request):
if viewer.short_id != user_short_id:
return Response("You are not authorized to delete this media.", status=403)
user_dbsession = request.user_dbsession # Assume this exists for verified users
user_dbsession = request.user_dbsession
media = user_dbsession.query(Media).filter_by(short_id=media_short_id).first()
if not media:
return Response("Media not found.", status=404)
# Delete the media
user_dbsession.delete(media)
user_dbsession.flush()
# No need to commit; transaction manager will handle it
return HTTPFound(
location=request.route_url("user_media", user_short_id=viewer.short_id)
@ -819,8 +845,7 @@ def edit_media_get_view(request):
if viewer.short_id != user_short_id:
return Response("You are not authorized to edit this media.", status=403)
user_dbsession = request.user_dbsession # Assume this exists for verified users
user_dbsession = request.user_dbsession
media = user_dbsession.query(Media).filter_by(short_id=media_short_id).first()
if not media:
return Response("Media not found.", status=404)
@ -843,8 +868,7 @@ def edit_media_post_view(request):
if viewer.short_id != user_short_id:
return Response("You are not authorized to edit this media.", status=403)
user_dbsession = request.user_dbsession # Assume this exists for verified users
user_dbsession = request.user_dbsession
media = user_dbsession.query(Media).filter_by(short_id=media_short_id).first()
if not media:
return Response("Media not found.", status=404)
@ -857,7 +881,7 @@ def edit_media_post_view(request):
new_media_file = request.POST.get("media_file")
if new_media_file and getattr(new_media_file, "filename", "").strip():
raw_bytes = new_media_file.file.read()
max_size = 30 * 1024 * 1024 # 30 MB
max_size = 30 * 1024 * 1024
if len(raw_bytes) > max_size:
return Response("File size exceeds the 30MB limit.", status=400)
file_size = len(raw_bytes)
@ -868,7 +892,6 @@ def edit_media_post_view(request):
media_type = mime_type.split("/")[0]
encoded_str = base64.b64encode(raw_bytes).decode("utf-8")
# Update media fields
media.filename = filename
media.media_type = media_type
media.media_b64 = encoded_str
@ -879,7 +902,6 @@ def edit_media_post_view(request):
media.is_public = is_public
user_dbsession.flush()
# No need to commit; transaction manager will handle it
return HTTPFound(
location=request.route_url(
@ -908,40 +930,37 @@ def view_media_view(request):
return Response("User not found.", status=404)
user_id = user.id
# Get user_dbsession
user_dbsession = get_user_dbsession_by_user_id(user_id, request)
if not user_dbsession:
return Response("User database not found.", status=404)
# Lookup media by short_id
media = user_dbsession.query(Media).filter_by(short_id=media_short_id).first()
if not media:
return Response("Media not found.", status=404)
viewer = request.user
is_owner = viewer and viewer.id == user.id
is_owner = viewer and viewer.id == user_id
if not media.is_public and not is_owner:
return Response("Media not available.", status=403)
# Decode base64 content
media_data = base64.b64decode(media.media_b64)
mime_type = get_mime_type(media.filename)
# Prepare filename for download
# Build a safe filename (avoid Unicode issues in the header)
if media.title:
# Use title as filename, slugify it
file_extension = os.path.splitext(media.filename)[1]
download_filename = f"{slugify(media.title)}{file_extension}"
raw_title = media.title
download_filename = sanitize_filename_for_http_header(
f"{raw_title}{file_extension}"
)
else:
# Use original filename
download_filename = media.filename
download_filename = sanitize_filename_for_http_header(media.filename)
# Check if the user wants to download the file
# Check if user wants attachment or inline
download = request.GET.get("download", "false").lower() == "true"
content_disposition = "attachment" if download else "inline"
# Serve the media content with appropriate headers
response = Response(body=media_data, content_type=mime_type)
response.headers.update(
{
@ -962,9 +981,9 @@ def main(global_config=None, **settings):
# Configure logging
logging.basicConfig(level=logging.DEBUG)
# Set up the session factory with the specified settings
# Set up the session factory
session_factory = SignedCookieSessionFactory(
secret="it-is-a-secret-you-must-change",
secret=pyrafiles_secret,
hashalg="sha512",
timeout=31104000, # Approx. one year in seconds
max_age=31104000, # Set Max-Age attribute on cookie
@ -983,7 +1002,6 @@ def main(global_config=None, **settings):
config.include("pyramid_tm") # Include pyramid_tm for transaction management
# Add .html.j2 extension for Jinja2 templates
# Set up Jinja2 template search path
config.add_jinja2_renderer(".j2")
config.add_jinja2_search_path("templates", name=".j2")
@ -993,14 +1011,13 @@ def main(global_config=None, **settings):
engine = create_engine(
DB_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool
)
session_factory = sessionmaker(bind=engine)
session_factory_ = sessionmaker(bind=engine)
Base.metadata.bind = engine
# Use a scoped_session
DBSession = scoped_session(session_factory)
DBSession = scoped_session(session_factory_)
register(DBSession) # Register with zope.sqlalchemy
# Add user to all requests.
# Add user to all requests
config.add_request_method(callable=get_current_user, name="user", reify=True)
# Provide dbsession to requests
@ -1025,10 +1042,10 @@ def main(global_config=None, **settings):
config.add_route("export_user_record", "/auth/export_user_record")
config.add_route("import_user_record", "/admin/import_user_record")
# Media - Reordered routes with most specific first
# Media
config.add_route("upload_media", "/media/upload")
config.add_route("list_media", "/media/list")
config.add_route("user_media", "/media/user/{user_short_id}") # Moved earlier
config.add_route("user_media", "/media/user/{user_short_id}")
config.add_route(
"view_media_details", "/media/{user_short_id}/{media_short_id}/details"
)
@ -1045,5 +1062,5 @@ def main(global_config=None, **settings):
if __name__ == "__main__":
app = main()
log.info("Serving on http://localhost:6544")
serve(app, host="0.0.0.0", port=6544)
log.info(f"Serving on http://{HOST}:{PORT}")
serve(app, host=HOST, port=PORT)