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

working with revocations

Browse source 
modified:   README.rst
	modified:   app.py
	modified:   templates/manage_namespace.html.j2
	modified:   test_agent.sh
This commit is contained in:
Russell Ballestrini 2025-01-11 20:18:34 -05:00
parent bed3607168
commit a8b0ee313d
4 changed files with 319 additions and 404 deletions
Download patch file Download diff file Expand all files Collapse all files

335
app.py
View file

@ -1,5 +1,6 @@
###############################################################################
# app.py - PyraLogs Application with JWT-based Authentication and Refresh Mechanism
# app.py - PyraLogs Application with Agent-Specific JWT Authentication and Revocation
# Users do not have token_version as only agents use JWTs
###############################################################################
import os
import base64
@ -149,36 +150,8 @@ def get_current_user(request):
# User ID in session does not exist in the database; remove it
del request.session["user_id"]
# No valid user in session; create a guest user
# Generate a unique guest username
while True:
suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=8))
guest_username = f"Guest-{suffix}"
existing_user = s.query(User).filter_by(username=guest_username).first()
if not existing_user:
break # Unique username found
# Generate UUID and short ID
user_uuid = uuid.uuid4()
new_user_id = str(user_uuid)
short_id = uuid_to_short_id(user_uuid)
# Create a new guest user
guest_user = User(
id=new_user_id,
short_id=short_id,
email=None, # Guests don't have an email
username=guest_username,
is_verified=False,
)
s.add(guest_user)
s.flush() # Use flush instead of commit in pyramid_tm
# Store the user ID in the session
request.session["user_id"] = guest_user.id
guest_user.dbsession = s # Attach dbsession to user
return guest_user
# No valid user in session; return None
return None
def uuid_to_short_id(u):
@ -226,12 +199,12 @@ Base = declarative_base()
# Association table for Namespace <-> User (with roles)
namespace_user_association = Table(
'namespace_user_association',
"namespace_user_association",
Base.metadata,
Column('namespace_id', String, ForeignKey('namespaces.id')),
Column('user_id', String, ForeignKey('users.id')),
Column('role', String, nullable=False), # 'owner', 'editor', 'reader'
Index('ix_namespace_user', 'namespace_id', 'user_id', unique=True)
Column("namespace_id", String, ForeignKey("namespaces.id")),
Column("user_id", String, ForeignKey("users.id")),
Column("role", String, nullable=False), # 'owner', 'editor', 'reader'
Index("ix_namespace_user", "namespace_id", "user_id", unique=True),
)
@ -239,8 +212,8 @@ 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) # Guests don't have an email.
username = Column(String, unique=True, nullable=False) # user-chosen handle
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)
@ -275,10 +248,27 @@ class Namespace(Base):
back_populates="namespaces",
)
# Agents associated with the namespace
agents = relationship("Agent", back_populates="namespace")
def __repr__(self):
return f"<Namespace(name='{self.name}', is_public={self.is_public})>"
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)
namespace = relationship("Namespace", back_populates="agents")
def __repr__(self):
return f"<Agent(name='{self.name}', namespace_id='{self.namespace_id}')>"
################################################################################
# Jinja2 Environment and Custom Filters
################################################################################
@ -430,11 +420,33 @@ class LogEntry(NamespaceBase):
@view_config(route_name="home", renderer="home.html.j2")
def home_view(request):
s = request.dbsession
# Show public namespaces
# 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 ns in request.user.namespaces:
result = s.execute(
namespace_user_association.select().where(
namespace_user_association.c.namespace_id == ns.id,
namespace_user_association.c.user_id == request.user.id,
)
).first()
if result:
user_namespaces.append(
{
"name": ns.name,
"short_id": ns.short_id,
"role": result.role,
}
)
return {
"request": request,
"public_namespaces": public_namespaces,
"user_namespaces": user_namespaces,
}
@ -553,14 +565,15 @@ def logout_view(request):
# JWT Helper Functions
################################################################################
def generate_jwt_token(user, expires_delta=datetime.timedelta(days=365)):
"""Generate a JWT for the given user."""
def generate_jwt_token(agent):
"""Generate a JWT for the given agent without an expiration time."""
payload = {
"user_id": user.id,
"username": user.username,
# Set 'iat' to issue time and 'nbf' to tomorrow to enforce daily refresh
"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,
# "exp": datetime.datetime.utcnow() + expires_delta, # Removed to make token not expire
}
token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
return token
@ -569,30 +582,17 @@ def generate_jwt_token(user, expires_delta=datetime.timedelta(days=365)):
def verify_jwt_token(token):
"""Verify the JWT and return the payload if valid."""
try:
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
payload = jwt.decode(
token,
JWT_SECRET,
algorithms=[JWT_ALGORITHM],
options={"verify_exp": False}, # Disable expiration verification
)
return payload
except PyJWTError:
return None
def issue_new_jwt_if_needed(old_token, user):
"""Issue a new JWT if the token was last used on a previous day."""
payload = verify_jwt_token(old_token)
if not payload:
return None # Invalid token
last_used = datetime.datetime.utcfromtimestamp(payload.get("last_used", 0))
now = datetime.datetime.utcnow()
if last_used.date() < now.date():
# Update 'last_used' to now and issue a new token
new_payload = payload.copy()
new_payload["last_used"] = now.timestamp()
token = jwt.encode(new_payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
return token
else:
return None # No need to issue a new token
################################################################################
# Profile and Namespace Management
################################################################################
@ -615,7 +615,7 @@ def profile_get_view(request):
namespace_user_association.c.user_id == user.id,
)
).first()
if result and result.role == 'owner':
if result and result.role == "owner":
owner_namespaces.append(ns)
return {
@ -651,7 +651,11 @@ def profile_post_view(request):
return HTTPFound(location=request.route_url("profile"))
@view_config(route_name="create_namespace", request_method="GET", renderer="create_namespace.html.j2")
@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)
@ -694,7 +698,7 @@ def create_namespace_post_view(request):
namespace_user_association.insert().values(
namespace_id=namespace.id,
user_id=user.id,
role='owner',
role="owner",
)
)
s.flush()
@ -702,12 +706,18 @@ def create_namespace_post_view(request):
# 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
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))
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")
@ -727,15 +737,21 @@ def manage_namespace_view(request):
users = []
for ur in user_roles:
u = s.query(User).filter(User.id == ur.user_id).first()
users.append({
'user': u,
'role': ur.role,
})
users.append(
{
"user": u,
"role": ur.role,
}
)
# Get agents associated with the namespace
agents = s.query(Agent).filter(Agent.namespace_id == namespace.id).all()
return {
"request": request,
"namespace": namespace,
"users": users,
"agents": agents,
}
@ -750,7 +766,11 @@ def update_namespace_view(request):
namespace.is_public = is_public
s.flush()
return HTTPFound(location=request.route_url("manage_namespace", namespace_short_id=namespace.short_id))
return HTTPFound(
location=request.route_url(
"manage_namespace", namespace_short_id=namespace.short_id
)
)
@view_config(route_name="invite_user", request_method="POST")
@ -761,7 +781,7 @@ def invite_user_view(request):
email = request.POST.get("email", "").strip().lower()
role = request.POST.get("role", "").strip().lower()
if role not in ['owner', 'editor', 'reader']:
if role not in ["owner", "editor", "reader"]:
return Response("Invalid role.", status=400)
# Find or create the user
@ -803,47 +823,44 @@ def invite_user_view(request):
email_body = f"You have been invited as a {role} to namespace '{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))
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")
@view_config(
route_name="generate_agent_jwt",
request_method="POST",
renderer="display_agent_jwt.html.j2",
)
@owner_required
def generate_agent_jwt_view(request):
namespace = request.namespace
user = request.user
s = request.dbsession
agent_name = request.POST.get("agent_name", "").strip()
if not agent_name:
return Response("Agent name is required.", status=400)
# Create a user entry for the agent without an email
# Generate UUID for the agent
agent_uuid = uuid.uuid4()
agent_id = str(agent_uuid)
agent_short_id = uuid_to_short_id(agent_uuid)
agent_user = User(
# Create an agent entry
agent = Agent(
id=agent_id,
short_id=agent_short_id,
email=None,
username=agent_name,
is_verified=True, # Agents are verified
name=agent_name,
namespace_id=namespace.id,
token_version=0,
)
s = request.dbsession
s.add(agent_user)
s.add(agent)
s.flush()
# Assign the agent to the namespace with the appropriate role
s.execute(
namespace_user_association.insert().values(
namespace_id=namespace.id,
user_id=agent_user.id,
role='editor', # Agents can be given 'editor' role
)
)
s.flush()
# Generate JWT for the agent
jwt_token = generate_jwt_token(agent_user)
# Generate JWT for the agent including the namespace_id
jwt_token = generate_jwt_token(agent)
return {
"request": request,
@ -852,6 +869,38 @@ def generate_agent_jwt_view(request):
"jwt_token": jwt_token,
}
@view_config(route_name="revoke_agent", request_method="POST")
@owner_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)
# Increment the agent's token_version to revoke its token
agent.token_version += 1
s.flush()
request.session.flash(f"Access revoked for agent '{agent.name}'.")
return HTTPFound(
location=request.route_url(
"manage_namespace", namespace_short_id=namespace.short_id
)
)
################################################################################
# Logging Views
################################################################################
@ -865,47 +914,48 @@ def log_webhook_view(request):
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()
token = auth_header[len("Bearer ") :].strip()
payload = verify_jwt_token(token)
if not payload:
return Response("Invalid or expired token.", status=401)
# Get user and check permissions
user = s.query(User).filter(User.id == payload["user_id"]).first()
if not user:
return Response("User not found.", status=404)
request.user = user
user.dbsession = s # Attach dbsession to user
# 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)
# Issue a new JWT if needed and include it in the response
new_token = issue_new_jwt_if_needed(token, user)
if new_token:
request.response.headers['X-New-JWT'] = new_token
# Check token_version
if payload.get("token_version") != agent.token_version:
return Response("Token has been revoked.", status=401)
# Get namespace from header or parameter
namespace_short_id = request.headers.get('X-Namespace') or request.params.get('namespace')
if not namespace_short_id:
return Response("Namespace not specified.", status=400)
namespace = s.query(Namespace).filter(Namespace.short_id == namespace_short_id).first()
# 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
# Check if user has sufficient permissions
if not check_namespace_permission(user, namespace, 'editor'):
return Response("You do not have permission to write to this namespace.", status=403)
request.namespace = namespace # Attach namespace to request
# No need to issue a new token since tokens do not expire due to time
# Access namespace_dbsession
namespace_dbsession = request.namespace_dbsession
payload = request.json_body
message = payload.get('message', '')
level = payload.get('level', 'INFO')
meta_data = payload.get('metadata', None)
if meta_data:
# 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,
@ -917,13 +967,19 @@ def log_webhook_view(request):
return Response("Log entry created.", status=201)
@view_config(route_name="view_namespace_logs", renderer="view_namespace_logs.html.j2")
@view_config(
route_name="view_namespace_logs", renderer="view_namespace_logs.html.j2"
)
@reader_required
def view_namespace_logs_view(request):
namespace = request.namespace
namespace_dbsession = request.namespace_dbsession
logs = namespace_dbsession.query(LogEntry).order_by(LogEntry.timestamp.desc()).all()
logs = (
namespace_dbsession.query(LogEntry)
.order_by(LogEntry.timestamp.desc())
.all()
)
return {
"request": request,
@ -969,11 +1025,13 @@ def main(global_config=None, **settings):
# Set up SQLAlchemy
engine = create_engine(
DB_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool
DB_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
session_factory_ = sessionmaker(bind=engine)
Base.metadata.bind = engine
#Base.metadata.create_all(engine)
Base.metadata.create_all(engine)
DBSession = scoped_session(session_factory_)
register(DBSession) # Register with zope.sqlalchemy
@ -991,7 +1049,9 @@ def main(global_config=None, **settings):
config.add_request_method(get_current_namespace, "namespace", reify=True)
# Add namespace_dbsession to requests if namespace is set
config.add_request_method(add_namespace_dbsession, "namespace_dbsession", reify=True)
config.add_request_method(
add_namespace_dbsession, "namespace_dbsession", reify=True
)
# Routes
config.add_route("home", "/")
@ -1004,15 +1064,24 @@ def main(global_config=None, **settings):
# Namespace Management
config.add_route("create_namespace", "/namespace/create")
config.add_route("manage_namespace", "/namespace/{namespace_short_id}/manage")
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(
"revoke_agent", "/namespace/{namespace_short_id}/revoke_agent"
)
# Route to generate JWTs for agents
config.add_route("generate_agent_jwt", "/namespace/{namespace_short_id}/generate_agent_jwt")
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.add_route(
"view_namespace_logs", "/namespace/{namespace_short_id}/logs"
)
config.scan()
return config.make_wsgi_app()