feat(jira): admin account replaces per-user tokens for all Jira/Tempo ops
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled

- Single admin Jira account stored in system_configs (jira.admin_email, jira.admin_api_token)
- Admin Tempo token in system_configs (tempo.admin_token)
- get_admin_jira_client() + has_admin_jira_configured() replace per-user auth in all lifecycle hooks
- lookup_user_jira_account_id() auto-discovers each user's Atlassian accountId by email on login
- auto_log_test_worklog() now uses admin Tempo token and logs both red+blue team time
- Settings > Jira: admin credentials fields + test buttons moved to admin Jira tab
- Profile section simplified: jira_account_id shown read-only (auto-detected)
This commit is contained in:
kitos
2026-06-24 09:15:35 +02:00
parent bb8b9a6a72
commit 18695197e8
7 changed files with 530 additions and 521 deletions
+109 -9
View File
@@ -188,10 +188,110 @@ def get_user_jira_client(user: User, db: Session):
def has_jira_configured(user: User, db: Session) -> bool:
"""Return True if *user* has everything needed to call Jira."""
"""Return True if *user* has everything needed to call Jira (legacy per-user check)."""
return bool(get_jira_url(db) and _effective_jira_email(user) and user.jira_api_token)
# ---------------------------------------------------------------------------
# Admin Jira client (single account for all lifecycle hooks)
# ---------------------------------------------------------------------------
def get_admin_jira_email(db: Session) -> Optional[str]:
"""Return the admin Jira email from system_configs."""
return _read_system_config(db, "jira.admin_email") or None
def get_admin_jira_api_token(db: Session) -> Optional[str]:
"""Return the admin Jira API token from system_configs."""
return _read_system_config(db, "jira.admin_api_token") or None
def has_admin_jira_configured(db: Session) -> bool:
"""Return True if the global Jira admin account is fully configured."""
return bool(
is_jira_enabled(db)
and get_jira_url(db)
and get_admin_jira_email(db)
and get_admin_jira_api_token(db)
)
def get_admin_jira_client(db: Session):
"""Return a Jira client authenticated with the global admin credentials.
All Aegis-to-Jira operations (issue creation, comments, transitions)
go through this single account. Users do not need personal Jira tokens.
"""
jira_url = get_jira_url(db)
if not jira_url:
raise InvalidOperationError("Jira URL is not configured.")
admin_email = get_admin_jira_email(db)
admin_token = get_admin_jira_api_token(db)
if not admin_email or not admin_token:
raise InvalidOperationError(
"Admin Jira credentials not configured. "
"Set them in System Settings → Jira Configuration → Admin Account."
)
from atlassian import Jira
return Jira(
url=jira_url.rstrip("/"),
username=admin_email,
password=admin_token,
cloud=True,
)
def lookup_user_jira_account_id(db: Session, user: User) -> bool:
"""Lookup *user*'s Atlassian account ID by email using the admin Jira client.
Updates ``user.jira_account_id`` in-place when found or changed.
Returns ``True`` when the value was updated, ``False`` otherwise.
Non-fatal — all errors are logged at DEBUG level and swallowed.
"""
if not has_admin_jira_configured(db):
return False
email = getattr(user, "email", None)
if not email:
return False
try:
jira = get_admin_jira_client(db)
results = jira.user_find_by_user_string(query=email, maxResults=10)
account_id: Optional[str] = None
for u in results or []:
if isinstance(u, dict) and u.get("emailAddress", "").lower() == email.lower():
account_id = u.get("accountId")
break
if not account_id:
logger.debug("No Jira user found for email %s", email)
return False
current = getattr(user, "jira_account_id", None)
if account_id != current:
user.jira_account_id = account_id
db.flush()
logger.info(
"Auto-updated jira_account_id for %s: %s", user.username, account_id
)
return True
return False
except Exception as exc:
logger.debug(
"Could not lookup Jira account_id for %s: %s",
getattr(user, "username", "?"),
exc,
)
return False
# ---------------------------------------------------------------------------
# Ticket content builders (inspired by the pentest-to-Jira script)
# ---------------------------------------------------------------------------
@@ -403,7 +503,7 @@ def auto_create_campaign_issue(
Called once right after a campaign is committed to the database.
The created ticket is stored as a JiraLink with entity_type=campaign.
"""
if not has_jira_configured(actor, db):
if not has_admin_jira_configured(db):
return None
project_key = get_jira_project_key(db)
@@ -417,7 +517,7 @@ def auto_create_campaign_issue(
parent_ticket = get_jira_parent_ticket(db)
try:
jira = get_user_jira_client(actor, db)
jira = get_admin_jira_client(db)
fields: dict = {
"project": {"key": project_key},
@@ -487,7 +587,7 @@ def auto_create_test_issue(
instead of the system-configured parent (e.g. OFS-9107).
Use this to nest test tickets under a campaign ticket.
"""
if not has_jira_configured(actor, db):
if not has_admin_jira_configured(db):
return None
project_key = get_jira_project_key(db)
@@ -502,7 +602,7 @@ def auto_create_test_issue(
mitre_id = technique.mitre_id if technique else "N/A"
try:
jira = get_user_jira_client(actor, db)
jira = get_admin_jira_client(db)
# All tests — whether inside a campaign or standalone — are created
# as Task. Campaign tests use the campaign Jira key as parent
@@ -574,7 +674,7 @@ def push_test_event(
Completely non-fatal — any Jira error is logged and swallowed so it
never blocks the test workflow.
"""
if not has_jira_configured(actor, db):
if not has_admin_jira_configured(db):
return
link = (
@@ -589,7 +689,7 @@ def push_test_event(
return
try:
jira = get_user_jira_client(actor, db)
jira = get_admin_jira_client(db)
comment = _build_state_comment(test, new_state, actor, extra)
jira.issue_add_comment(link.jira_issue_key, comment)
@@ -650,7 +750,7 @@ def push_hold_event(
Non-fatal — any Jira error is logged and swallowed.
"""
if not has_jira_configured(actor, db):
if not has_admin_jira_configured(db):
return
link = (
@@ -665,7 +765,7 @@ def push_hold_event(
return
try:
jira = get_user_jira_client(actor, db)
jira = get_admin_jira_client(db)
ts = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
if resuming: