############################################################################### # app.py - PyraLogs Application with Enhanced Secret Management # - Secrets are generated once, saved to the filesystem, and loaded thereafter # - Different secrets are used for session signing and JWT signing # - Includes previous enhancements for agent management and OpenAPI integration ############################################################################### import os import base64 import datetime import random import string import bcrypt import re import uuid import hashlib import smtplib import mimetypes import json import logging import unicodedata from email.mime.text import MIMEText from pyramid.config import Configurator from pyramid.view import view_config from pyramid.response import Response from pyramid.httpexceptions import HTTPFound, HTTPForbidden, HTTPNotFound from pyramid.session import SignedCookieSessionFactory from sqlalchemy import ( create_engine, Column, String, DateTime, Boolean, Integer, Index, ForeignKey, Text, or_, ) from sqlalchemy.orm import declarative_base from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import scoped_session, relationship from sqlalchemy.pool import StaticPool from waitress import serve from pyramid.renderers import render_to_response from pyramid.events import subscriber from pyramid_jinja2 import IJinja2Environment import transaction # Import transaction management from zope.sqlalchemy import register # Import zope.sqlalchemy import jwt # Import PyJWT library from jwt import PyJWTError ################################################################################ # Set up logging ################################################################################ logging.basicConfig(level=logging.INFO) log = logging.getLogger(__name__) ################################################################################ # Environment Variables and Defaults ################################################################################ # Application directory APP_DIR = os.path.dirname(os.path.abspath(__file__)) # Data directory DATA_DIR = os.path.join(APP_DIR, "data") # Ensure DATA_DIR exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Paths to the secret files PYRALOGS_SECRET_FILE = os.path.join(DATA_DIR, "pyralogs_secret.txt") JWT_SECRET_FILE = os.path.join(DATA_DIR, "jwt_secret.txt") def get_or_create_secret(env_var_name, secret_file_path): """ Retrieves the secret from an environment variable, or loads it from the specified file. If neither is available, generates a new secret, saves it to the file, and returns it. """ # Check environment variable secret = os.environ.get(env_var_name, "") if secret: log.info(f"Using {env_var_name} from environment variable.") return secret # Check if the secret file exists if os.path.exists(secret_file_path): with open(secret_file_path, "r") as f: secret = f.read().strip() if secret: log.info(f"Loaded {env_var_name} from {secret_file_path}") return secret else: log.warning(f"{secret_file_path} is empty. Generating new secret.") else: log.info(f"{secret_file_path} does not exist. Generating new secret.") # Generate a new secret secret = "".join(random.choices(string.ascii_letters + string.digits, k=64)) # Save the secret to the file with open(secret_file_path, "w") as f: f.write(secret) log.info(f"Generated and saved new {env_var_name} to {secret_file_path}") return secret # Retrieve or generate the secrets pyralogs_secret = get_or_create_secret("PYRALOGS_SECRET", PYRALOGS_SECRET_FILE) JWT_SECRET = get_or_create_secret("PYRALOGS_JWT_SECRET", JWT_SECRET_FILE) JWT_ALGORITHM = "HS256" # Database URL can be overridden by environment variable default_main_db_url = f"sqlite:///{os.path.join(DATA_DIR, 'main.db')}" DB_URL = os.environ.get("PYRALOGS_DB_URL", default_main_db_url) # Host and port for the application HOST = os.environ.get("PYRALOGS_HOST", "0.0.0.0") PORT = int(os.environ.get("PYRALOGS_PORT", "6544")) # SMTP host/port smtp_host = os.environ.get("PYRALOGS_SMTP_HOST", "localhost") smtp_port = int(os.environ.get("PYRALOGS_SMTP_PORT", "25")) ################################################################################ # Helper Functions ################################################################################ def slugify(text): text = text.lower() text = re.sub(r"\s+", "-", text) text = re.sub(r"[^\w\-]", "", text) return text def get_gravatar_url(email, size=100): email = email.strip().lower() hash_code = hashlib.md5(email.encode("utf-8")).hexdigest() return f"https://www.gravatar.com/avatar/{hash_code}?s={size}&d=identicon" def send_email(to_email, subject, body, from_email=None): if from_email is None: from_email = os.environ.get("PYRALOGS_FROM_EMAIL", "master@master.unturf.com") msg = MIMEText(body) msg["Subject"] = subject msg["From"] = from_email msg["To"] = to_email try: s = smtplib.SMTP(smtp_host, smtp_port) s.sendmail(from_email, [to_email], msg.as_string()) s.quit() except Exception as e: log = logging.getLogger(__name__) log.info("======= Email Sent =======") log.info(f"To: {to_email}") log.info(f"Subject: {subject}") log.info(f"Body:\n{body}") log.info("==========================") log.error(f"Error sending email: {e}") def admin_required(view_func): def wrapper(request): user = request.user if not user or not user.is_admin: return HTTPForbidden("You must be an admin to access this page.") return view_func(request) return wrapper def get_current_user(request): """Return the current user (authenticated or guest) from session.""" user_id = request.session.get("user_id") s = request.dbsession if user_id: # Try to get the user from the database user = s.query(User).filter_by(id=user_id).first() if user: user.dbsession = s # Attach dbsession to user return user else: # User ID in session does not exist in the database; remove it del request.session["user_id"] # No valid user in session; return None return None def get_user_namespace_role(request): user = request.user namespace = request.namespace if not user or not namespace: return None s = request.dbsession association = ( s.query(NamespaceUserAssociation) .filter( NamespaceUserAssociation.user_id == user.id, NamespaceUserAssociation.namespace_id == namespace.id, ) .first() ) if association: return association.role else: return None def uuid_to_short_id(u): """Encode UUID to a URL-safe base64 string without padding.""" return base64.urlsafe_b64encode(u.bytes).decode("ascii").rstrip("=") def short_id_to_uuid(sid): """Decode the short ID back to UUID, trying different padding lengths.""" 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: continue log.error(f"Failed to convert short_id {sid} to UUID after trying all paddings") return None def get_namespace_db_url(namespace_id): """Return the database URL for the namespace's SQLite database.""" db_file = os.path.join(DATA_DIR, f"namespace_{namespace_id}.db") return f"sqlite:///{db_file}" def slugify_filename(filename): """Sanitize and slugify filename to prevent path traversal.""" filename = os.path.basename(filename) filename = re.sub(r"[^\w\.-]", "_", filename) return filename or "file" ################################################################################ # Database Setup ################################################################################ log.debug(f"Using database URL: {DB_URL}") # For debugging Base = declarative_base() # Association class for Namespace <-> User (with roles) class NamespaceUserAssociation(Base): __tablename__ = "namespace_user_association" namespace_id = Column(String, ForeignKey("namespaces.id"), primary_key=True) user_id = Column(String, ForeignKey("users.id"), primary_key=True) role = Column(String, nullable=False) # 'owner', 'editor', 'reader' namespace = relationship("Namespace", back_populates="user_associations") user = relationship("User", back_populates="namespace_associations") class User(Base): __tablename__ = "users" id = Column(String, primary_key=True) # UUID short_id = Column(String, unique=True, nullable=False) email = Column(String, unique=True, nullable=True) username = Column(String, unique=True, nullable=False) code_hash = Column(String, nullable=True) # bcrypt hash of code code_expires = Column(DateTime, nullable=True) # time limit for code is_verified = Column(Boolean, default=False) enable_gravatar = Column(Boolean, default=False) # Gravatar support is_admin = Column(Boolean, default=False) # Admin flag # Namespaces the user is associated with namespace_associations = relationship( "NamespaceUserAssociation", back_populates="user" ) namespaces = relationship( "Namespace", secondary="namespace_user_association", back_populates="users", ) # Attribute to hold dbsession in permission checks dbsession = None def __repr__(self): return f"" class Namespace(Base): __tablename__ = "namespaces" id = Column(String, primary_key=True) # UUID short_id = Column(String, unique=True, nullable=False) name = Column(String, unique=True, nullable=False) is_public = Column(Boolean, default=False) # Users associated with the namespace user_associations = relationship( "NamespaceUserAssociation", back_populates="namespace" ) users = relationship( "User", secondary="namespace_user_association", back_populates="namespaces", ) # Agents associated with the namespace agents = relationship("Agent", back_populates="namespace") def __repr__(self): return f"" class Agent(Base): __tablename__ = "agents" id = Column(String, primary_key=True) # UUID name = Column(String, nullable=False) namespace_id = Column(String, ForeignKey("namespaces.id")) token_version = Column(Integer, default=0) created_at = Column(DateTime, default=datetime.datetime.utcnow) status = Column(String, default="active") # 'active' or 'revoked' namespace = relationship("Namespace", back_populates="agents") def __repr__(self): return ( f"" ) ################################################################################ # Jinja2 Environment and Custom Filters ################################################################################ @subscriber(IJinja2Environment) def add_jinja2_filters(event): env = event.environment # Add any custom Jinja2 filters here ################################################################################ # Request Methods ################################################################################ def get_namespace(request): """Get the namespace from the route parameter 'namespace_short_id'.""" namespace_short_id = request.matchdict.get("namespace_short_id") if not namespace_short_id: return None s = request.dbsession namespace = s.query(Namespace).filter_by(short_id=namespace_short_id).first() return namespace def get_namespace_dbsession(request): """Adds namespace_dbsession to request if namespace is set.""" namespace = request.namespace if namespace: namespace_dbsession = get_namespace_dbsession_by_namespace_id( namespace.id, request ) return namespace_dbsession else: return None # No namespace selected def get_namespace_dbsession_by_namespace_id(namespace_id, request): """Helper function to get a namespace_dbsession for a given namespace_id.""" namespace_db_url = get_namespace_db_url(namespace_id) db_file = os.path.join(DATA_DIR, f"namespace_{namespace_id}.db") if not os.path.exists(db_file): # Create the namespace database if it doesn't exist engine = create_engine( namespace_db_url, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) NamespaceBase.metadata.create_all(engine) engine.dispose() namespace_engine = create_engine( namespace_db_url, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) NamespaceSessionFactory = sessionmaker(bind=namespace_engine) namespace_dbsession = scoped_session(NamespaceSessionFactory) register(namespace_dbsession) # Register with zope.sqlalchemy # Attach cleanup callbacks def cleanup(_request): namespace_dbsession.remove() namespace_engine.dispose() request.add_finished_callback(cleanup) return namespace_dbsession def check_namespace_permission(user, namespace, required_role): """Check if the user has the required role in the namespace.""" if not namespace: return False s = user.dbsession association = ( s.query(NamespaceUserAssociation) .filter( NamespaceUserAssociation.namespace_id == namespace.id, NamespaceUserAssociation.user_id == user.id, ) .first() ) if not association: if namespace.is_public and required_role == "reader": return True else: return False user_role = association.role roles_hierarchy = {"owner": 3, "editor": 2, "reader": 1} return roles_hierarchy.get(user_role, 0) >= roles_hierarchy.get(required_role, 0) def owner_required(view_func): def wrapper(request): user = request.user namespace = request.namespace if not namespace: return HTTPNotFound("Namespace not found.") if not user or not check_namespace_permission(user, namespace, "owner"): return HTTPForbidden("You must be an owner to access this page.") return view_func(request) return wrapper def editor_required(view_func): def wrapper(request): user = request.user namespace = request.namespace if not namespace: return HTTPNotFound("Namespace not found.") if not user or not check_namespace_permission(user, namespace, "editor"): return HTTPForbidden("You must be an editor to access this page.") return view_func(request) return wrapper def reader_required(view_func): def wrapper(request): user = request.user namespace = request.namespace if not namespace: return HTTPNotFound("Namespace not found.") if not user or not check_namespace_permission(user, namespace, "reader"): return HTTPForbidden("You do not have access to this namespace.") return view_func(request) return wrapper ################################################################################ # Namespace Database Models ################################################################################ NamespaceBase = declarative_base() class LogEntry(NamespaceBase): __tablename__ = "log_entries" id = Column(Integer, primary_key=True, autoincrement=True) timestamp = Column(DateTime, default=datetime.datetime.utcnow) message = Column(Text, nullable=False) level = Column(String, nullable=False) log_metadata = Column(Text, nullable=True) def __repr__(self): return f"" ################################################################################ # Routes and Views ################################################################################ @view_config(route_name="home", renderer="home.html.j2") def home_view(request): s = request.dbsession # Get public namespaces public_namespaces = s.query(Namespace).filter(Namespace.is_public == True).all() user_namespaces = [] if request.user and request.user.is_verified: # Get namespaces the user has access to along with their roles user_namespaces = [] for association in request.user.namespace_associations: ns = association.namespace user_namespaces.append( { "name": ns.name, "short_id": ns.short_id, "role": association.role, } ) return { "request": request, "public_namespaces": public_namespaces, "user_namespaces": user_namespaces, } ################################################################################ # Authentication Views ################################################################################ @view_config(route_name="login", request_method="GET", renderer="login.html.j2") def login_get_view(request): return {"request": request} @view_config(route_name="login", request_method="POST") def login_post_view(request): email = request.POST.get("email", "").strip().lower() if not email: return Response("Email required.", status=400) session = request.dbsession user = session.query(User).filter_by(email=email).first() if not user: # Generate UUID and short ID user_uuid = uuid.uuid4() user_id = str(user_uuid) short_id = uuid_to_short_id(user_uuid) # Create new user user = User( id=user_id, short_id=short_id, email=email, username=email.split("@")[0], is_verified=False, ) session.add(user) session.flush() # Generate 6-digit code code_str = f"{random.randint(0,999999):06d}" code_hash = bcrypt.hashpw(code_str.encode("utf-8"), bcrypt.gensalt()).decode( "utf-8" ) user.code_hash = code_hash user.code_expires = datetime.datetime.now() + datetime.timedelta(minutes=15) user.is_verified = False session.flush() # Send code via email email_body = f"Your verification code is: {code_str}" send_email(user.email, "Your Verification Code", email_body) # Store the email in the session for verification request.session["login_email"] = email return HTTPFound(location=request.route_url("verify")) @view_config(route_name="verify", request_method="GET", renderer="verify.html.j2") def verify_get_view(request): return {"request": request} @view_config(route_name="verify", request_method="POST") def verify_post_view(request): code_entered = request.POST.get("code", "").strip() if not code_entered or len(code_entered) != 6: return Response("Invalid code.", status=400) email = request.session.get("login_email") if not email: return Response( "No email found in session. Please start the login process again.", status=400, ) s = request.dbsession user = ( s.query(User) .filter( User.email == email, User.code_expires > datetime.datetime.now(), User.code_hash != None, ) .first() ) if not user: return Response("Code not found or expired.", status=400) if not bcrypt.checkpw(code_entered.encode("utf-8"), user.code_hash.encode("utf-8")): return Response("Invalid code.", status=400) user.is_verified = True user.code_hash = None user.code_expires = None s.flush() # Remove the email from the session del request.session["login_email"] request.session["user_id"] = user.id # Authentication cookies are used for other routes. return HTTPFound(location=request.route_url("home")) @view_config(route_name="logout", request_method="POST", require_csrf=True) def logout_view(request): request.session.invalidate() return HTTPFound(location=request.route_url("home")) ################################################################################ # JWT Helper Functions ################################################################################ def generate_jwt_token(agent): """Generate a JWT for the given agent without an expiration time.""" payload = { "agent_id": agent.id, "agent_name": agent.name, "namespace_id": agent.namespace_id, "token_version": agent.token_version, "iat": datetime.datetime.utcnow(), # "exp": datetime.datetime.utcnow() + expires_delta, # Removed to make token not expire } token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) return token def verify_jwt_token(token): """Verify the JWT and return the payload if valid.""" try: payload = jwt.decode( token, JWT_SECRET, algorithms=[JWT_ALGORITHM], options={"verify_exp": False}, # Disable expiration verification ) return payload except PyJWTError: return None ################################################################################ # Profile and Namespace Management ################################################################################ @view_config(route_name="profile", request_method="GET", renderer="profile.html.j2") def profile_get_view(request): user = request.user if not user: return Response("You must be logged in to access your profile.", status=403) gravatar_url = get_gravatar_url(user.email) if user.enable_gravatar else "" # Get namespaces where the user is an owner owner_namespaces = [] for association in user.namespace_associations: if association.role == "owner": owner_namespaces.append(association.namespace) return { "request": request, "user": user, "gravatar_url": gravatar_url, "owner_namespaces": owner_namespaces, } @view_config(route_name="profile", request_method="POST") def profile_post_view(request): if not request.user: return Response("You must be logged in to update your profile.", status=403) if not request.user.is_verified: return Response( "This account is in guest mode, log in to update your profile.", status=403, ) s = request.dbsession enable_gravatar = request.POST.get("enable_gravatar") == "on" request.user.enable_gravatar = enable_gravatar # Handle username update if provided new_username = request.POST.get("new_username", "").strip() if new_username: # Check if the new username is already taken existing = s.query(User).filter(User.username == new_username).first() if existing and existing.id != request.user.id: return Response("Username is already in use.", status=400) request.user.username = new_username s.flush() return HTTPFound(location=request.route_url("profile")) @view_config( route_name="create_namespace", request_method="GET", renderer="create_namespace.html.j2", ) def create_namespace_get_view(request): if not request.user or not request.user.is_verified: return Response("You must be logged in to create a namespace.", status=403) return {"request": request} @view_config(route_name="create_namespace", request_method="POST") def create_namespace_post_view(request): user = request.user if not user or not user.is_verified: return Response("You must be logged in to create a namespace.", status=403) name = request.POST.get("name", "").strip() if not name: return Response("Namespace name is required.", status=400) s = request.dbsession existing_namespace = s.query(Namespace).filter(Namespace.name == name).first() if existing_namespace: return Response("Namespace name already exists.", status=400) # Generate UUID and short ID for the namespace namespace_uuid = uuid.uuid4() namespace_id = str(namespace_uuid) namespace_short_id = uuid_to_short_id(namespace_uuid) is_public = request.POST.get("is_public") == "on" namespace = Namespace( id=namespace_id, short_id=namespace_short_id, name=name, is_public=is_public, ) s.add(namespace) # Add the user as an owner association = NamespaceUserAssociation( namespace=namespace, user=user, role="owner", ) s.add(association) s.flush() # Create namespace database namespace_db_url = get_namespace_db_url(namespace.id) engine = create_engine( namespace_db_url, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) NamespaceBase.metadata.create_all(engine) engine.dispose() return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) @view_config(route_name="manage_namespace", renderer="manage_namespace.html.j2") @editor_required def manage_namespace_view(request): namespace = request.namespace s = request.dbsession # Get users and their roles in the namespace associations = ( s.query(NamespaceUserAssociation) .filter(NamespaceUserAssociation.namespace_id == namespace.id) .all() ) users = [] for association in associations: user = association.user users.append({"user": user, "role": association.role}) # Get agents associated with the namespace (only active agents) agents = ( s.query(Agent) .filter(Agent.namespace_id == namespace.id, Agent.status == "active") .all() ) return { "request": request, "namespace": namespace, "users": users, "agents": agents, } @view_config(route_name="change_member_role", request_method="POST") @owner_required def change_member_role_view(request): s = request.dbsession namespace = request.namespace current_user = request.user user_id = request.POST.get("user_id") new_role = request.POST.get("role") if not user_id or not new_role: return Response("User ID and new role are required.", status=400) if new_role not in ["owner", "editor", "reader"]: return Response("Invalid role.", status=400) # Prevent owners from changing their own role if user_id == current_user.id: return Response("Owners cannot change their own role.", status=400) # Ensure the user is a member of the namespace association = ( s.query(NamespaceUserAssociation) .filter( NamespaceUserAssociation.namespace_id == namespace.id, NamespaceUserAssociation.user_id == user_id, ) .first() ) if not association: return Response("User is not a member of this namespace.", status=400) # Update the user's role association.role = new_role s.flush() request.session.flash(f"User's role has been updated to {new_role}.") return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) @view_config(route_name="update_namespace", request_method="POST") @owner_required def update_namespace_view(request): namespace = request.namespace s = request.dbsession # Update namespace properties is_public = request.POST.get("is_public") == "on" namespace.is_public = is_public s.flush() return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) @view_config(route_name="invite_user", request_method="POST") @owner_required def invite_user_view(request): namespace = request.namespace s = request.dbsession email = request.POST.get("email", "").strip().lower() role = request.POST.get("role", "").strip().lower() if role not in ["owner", "editor", "reader"]: return Response("Invalid role.", status=400) # Find or create the user user = s.query(User).filter(User.email == email).first() if not user: user_uuid = uuid.uuid4() user_id = str(user_uuid) short_id = uuid_to_short_id(user_uuid) user = User( id=user_id, short_id=short_id, email=email, username=email.split("@")[0], is_verified=False, ) s.add(user) s.flush() # Check if the user already has an association with the namespace existing_association = ( s.query(NamespaceUserAssociation) .filter( NamespaceUserAssociation.namespace_id == namespace.id, NamespaceUserAssociation.user_id == user.id, ) .first() ) if existing_association: # Update the role if the user is already associated existing_association.role = role else: # Create a new association association = NamespaceUserAssociation( namespace=namespace, user=user, role=role, ) s.add(association) s.flush() request.session.flash(f"{user.email} was invited as {role} role.") # Send invitation email email_body = ( f"You have been invited as a {role} to namespace " f"'{namespace.name}'. Please log in to access it." ) send_email(user.email, "Namespace Invitation", email_body) return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) @view_config(route_name="remove_user", request_method="POST") @owner_required def remove_user_view(request): s = request.dbsession namespace = request.namespace current_user = request.user # The owner initiating the removal # Get the user ID to remove from the POST data user_id_to_remove = request.POST.get("user_id") if not user_id_to_remove: return Response("User ID is required.", status=400) # Ensure that the user exists and is a member of the namespace user_to_remove = s.query(User).filter(User.id == user_id_to_remove).first() if not user_to_remove: return Response("User not found.", status=404) # Prevent owners from removing themselves if user_to_remove.id == current_user.id: return Response("Owners cannot remove themselves.", status=400) # Check if the user to remove is a member of the namespace association = ( s.query(NamespaceUserAssociation) .filter( NamespaceUserAssociation.namespace_id == namespace.id, NamespaceUserAssociation.user_id == user_id_to_remove, ) .first() ) if not association: return Response("User is not a member of this namespace.", status=400) # Remove the association s.delete(association) s.flush() # Provide a success message request.session.flash( f"User '{user_to_remove.username}' has been removed from the namespace." ) return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) @view_config( route_name="generate_agent_jwt", request_method="POST", renderer="display_agent_jwt.html.j2", ) @editor_required def generate_agent_jwt_view(request): namespace = request.namespace s = request.dbsession agent_name = request.POST.get("agent_name", "").strip() if not agent_name: return Response("Agent name is required.", status=400) # Check if an agent with the same name exists in the namespace agent = ( s.query(Agent) .filter( Agent.name == agent_name, Agent.namespace_id == namespace.id, ) .first() ) if agent: # Agent exists, increment token_version and regenerate JWT agent.token_version += 1 agent.status = "active" # Ensure the agent is active s.flush() jwt_token = generate_jwt_token(agent) message = ( f"A new JWT has been generated for existing agent '{agent.name}'. " "Any previous tokens have been revoked." ) else: # Generate UUID for the new agent agent_uuid = uuid.uuid4() agent_id = str(agent_uuid) # Create a new agent entry agent = Agent( id=agent_id, name=agent_name, namespace_id=namespace.id, token_version=0, status="active", ) s.add(agent) s.flush() # Generate JWT for the agent including the namespace_id jwt_token = generate_jwt_token(agent) message = f"A new agent '{agent.name}' has been created." return { "request": request, "namespace": namespace, "agent_name": agent_name, "jwt_token": jwt_token, "message": message, } @view_config(route_name="revoke_agent", request_method="POST") @editor_required def revoke_agent_view(request): namespace = request.namespace s = request.dbsession agent_id = request.POST.get("agent_id") if not agent_id: return Response("Agent ID is required.", status=400) # Get the agent agent = ( s.query(Agent) .filter( Agent.id == agent_id, Agent.namespace_id == namespace.id, ) .first() ) if not agent: return Response("Agent not found.", status=404) # Set the agent's status to 'revoked' to hide it from the dashboard agent.status = "revoked" s.flush() request.session.flash(f"Agent '{agent.name}' has been revoked.") return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) ################################################################################ # Logging Views ################################################################################ @view_config(route_name="log_webhook", request_method="POST") def log_webhook_view(request): s = request.dbsession # Attempt authentication via JWT auth_header = request.headers.get("Authorization") if not auth_header or not auth_header.startswith("Bearer "): return Response("Missing or invalid Authorization header.", status=401) token = auth_header[len("Bearer ") :].strip() payload = verify_jwt_token(token) if not payload: return Response("Invalid or expired token.", status=401) # Get agent from the JWT agent_id = payload.get("agent_id") if not agent_id: return Response("Agent ID not specified in token.", status=400) agent = s.query(Agent).filter(Agent.id == agent_id).first() if not agent: return Response("Agent not found.", status=404) # Check agent status if agent.status != "active": return Response("Agent is revoked or inactive.", status=401) # Check token_version if payload.get("token_version") != agent.token_version: return Response("Token has been revoked.", status=401) # Get namespace from the agent namespace = s.query(Namespace).filter(Namespace.id == agent.namespace_id).first() if not namespace: return Response("Namespace not found.", status=404) request.namespace = namespace # No need to issue a new token since tokens do not expire due to time # Access namespace_dbsession namespace_dbsession = request.namespace_dbsession # Parse payload try: json_payload = request.json_body except json.decoder.JSONDecodeError: return Response("Invalid JSON payload.", status=400) message = json_payload.get("message", "") if not message: return Response("Log message is required.", status=400) level = json_payload.get("level", "INFO") # Default to 'INFO' if not provided meta_data = json_payload.get("metadata", None) if meta_data is not None: meta_data = json.dumps(meta_data) # Create log entry log_entry = LogEntry( message=message, level=level, log_metadata=meta_data, ) namespace_dbsession.add(log_entry) namespace_dbsession.flush() return Response("Log entry created.", status=201) @view_config(route_name="view_namespace_logs", renderer="view_namespace_logs.html.j2") def view_namespace_logs_view(request): namespace = request.namespace if not namespace: return HTTPNotFound("Namespace not found.") namespace_dbsession = request.namespace_dbsession # Check if the namespace is public if namespace.is_public: # Allow access to anyone pass # Proceed to retrieve logs else: # Namespace is private # Check if user is authenticated user = request.user if not user: # User is not authenticated return HTTPForbidden("You must be logged in to access this namespace.") # Check if user has at least 'reader' role in this namespace if not check_namespace_permission(user, namespace, required_role="reader"): return HTTPForbidden("You do not have access to this namespace.") # Get the search query from the request parameters search_query = request.params.get("q", "").strip() # Start building the query query = namespace_dbsession.query(LogEntry) if search_query: # Create a search pattern for case-insensitive search search_pattern = f"%{search_query}%" query = query.filter( or_( LogEntry.message.ilike(search_pattern), LogEntry.level.ilike(search_pattern), LogEntry.log_metadata.ilike(search_pattern), ) ) logs = query.order_by(LogEntry.timestamp.desc()).all() return { "request": request, "namespace": namespace, "logs": logs, "search_query": search_query, } ################################################################################ # Main ################################################################################ def main(*config, **settings): # Configure logging logging.basicConfig(level=logging.INFO) # Set up the session factory session_factory = SignedCookieSessionFactory( secret=pyralogs_secret, hashalg="sha512", timeout=31104000, # Approx. one year in seconds max_age=31104000, # Set Max-Age attribute on cookie reissue_time=15552000, # Approx. six months samesite=None, # Allows cross-site requests if needed httponly=True, # Helps mitigate XSS attacks secure=False, # Set to True if using HTTPS ) if not settings: settings = {} settings["sqlalchemy.url"] = DB_URL config = Configurator(settings=settings, session_factory=session_factory) config.include("pyramid_jinja2") # Include pyramid_tm for transaction management config.include("pyramid_tm") config.include("pyramid_openapi3") config.pyramid_openapi3_spec("openapi.yaml", route="/openapi.yaml") config.pyramid_openapi3_add_explorer(route="/docs") # Add .html.j2 extension for Jinja2 templates config.add_jinja2_renderer(".j2") config.add_jinja2_search_path("templates", name=".j2") # The Jinja2 filters are added via the event subscriber above # Set up SQLAlchemy engine = create_engine( DB_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) session_factory_ = sessionmaker(bind=engine) Base.metadata.bind = engine DBSession = scoped_session(session_factory_) # Register with zope.sqlalchemy register(DBSession) # Provide dbsession to requests def dbsession(request): return DBSession config.add_request_method(dbsession, "dbsession", reify=True) # Add user, namespace, user_namespace_role to requests config.add_request_method(get_current_user, "user", reify=True) config.add_request_method(get_namespace, "namespace", reify=True) config.add_request_method( get_user_namespace_role, "user_namespace_role", reify=True ) # Add namespace_dbsession to requests if namespace is set config.add_request_method( get_namespace_dbsession, "namespace_dbsession", reify=True ) # Routes config.add_route("home", "/") # Auth for cookie sessions. config.add_route("login", "/auth/login") config.add_route("verify", "/auth/verify") config.add_route("logout", "/auth/logout") config.add_route("profile", "/auth/profile") # Namespace Management config.add_route("create_namespace", "/namespace/create") config.add_route("manage_namespace", "/namespace/{namespace_short_id}/manage") config.add_route("update_namespace", "/namespace/{namespace_short_id}/update") config.add_route("invite_user", "/namespace/{namespace_short_id}/invite") config.add_route("remove_user", "/namespace/{namespace_short_id}/remove_user") config.add_route( "change_member_role", "/namespace/{namespace_short_id}/change_member_role" ) # Routes to generate JWTs for agents without access to an email mailbox. config.add_route("revoke_agent", "/namespace/{namespace_short_id}/revoke_agent") config.add_route( "generate_agent_jwt", "/namespace/{namespace_short_id}/generate_agent_jwt" ) # Logging config.add_route("log_webhook", "/webhook") config.add_route("view_namespace_logs", "/namespace/{namespace_short_id}/logs") config.scan() return config.make_wsgi_app() # Expose the WSGI application callable for uwsgi uwsgi_app = main({}) if __name__ == "__main__": app = main({}) log.info(f"Serving on http://{HOST}:{PORT}") serve(app, host=HOST, port=PORT)