from flask import Flask, request, render_template_string
import subprocess
import os
import urllib.request
import urllib.error
import ssl
import json
from datetime import datetime, timezone
app = Flask(__name__)
DEBUG_PATH = os.environ.get('DEBUG_PATH')
DEPLOYMENT_NAME = os.environ.get('DEPLOYMENT_NAME', 'phoenix-app')
DASHBOARD_HTML = '''
Phoenix
Phoenix
{% if error %}
{{ error }}
{% else %}
Phoenix is up and running for {{ uptime }}
Details
| Deployment | {{ deployment }} |
| Status | {{ status }} |
| Replicas | {{ replicas }} |
| Namespace | {{ namespace }} |
| Created | {{ created }} |
{% endif %}
'''
def get_deployment_info():
"""Fetch deployment info from K8s API using mounted ServiceAccount token."""
try:
with open('/var/run/secrets/kubernetes.io/serviceaccount/token') as f:
token = f.read().strip()
with open('/var/run/secrets/kubernetes.io/serviceaccount/namespace') as f:
namespace = f.read().strip()
ctx = ssl.create_default_context()
ctx.load_verify_locations('/var/run/secrets/kubernetes.io/serviceaccount/ca.crt')
api_url = f'https://kubernetes.default.svc/apis/apps/v1/namespaces/{namespace}/deployments/{DEPLOYMENT_NAME}'
req = urllib.request.Request(api_url, headers={'Authorization': f'Bearer {token}'})
with urllib.request.urlopen(req, context=ctx, timeout=5) as resp:
deploy = json.loads(resp.read())
created_str = deploy['metadata']['creationTimestamp']
created_dt = datetime.fromisoformat(created_str.replace('Z', '+00:00'))
now = datetime.now(timezone.utc)
uptime_delta = now - created_dt
days = uptime_delta.days
hours, remainder = divmod(uptime_delta.seconds, 3600)
minutes, _ = divmod(remainder, 60)
if days > 0:
uptime = f"{days}d {hours}h {minutes}m"
elif hours > 0:
uptime = f"{hours}h {minutes}m"
else:
uptime = f"{minutes}m"
ready = deploy.get('status', {}).get('readyReplicas', 0)
desired = deploy.get('spec', {}).get('replicas', 1)
status = 'Running' if ready == desired else 'Degraded'
return {
'deployment': DEPLOYMENT_NAME,
'namespace': namespace,
'status': status,
'replicas': f"{ready}/{desired}",
'created': created_dt.strftime('%Y-%m-%d %H:%M:%S UTC'),
'uptime': uptime,
'error': None
}
except FileNotFoundError:
return {'error': 'Not running in Kubernetes'}
except urllib.error.HTTPError as e:
if e.code == 403:
return {'error': 'Unable to fetch deployment details (forbidden)'}
return {'error': f'Unable to fetch deployment details ({e.code})'}
except Exception as e:
return {'error': str(e)}
HTML = '''
Debug Console
Phoenix Debug Console
Internal use only - remove before production deployment
{% if output %}
$ {{ cmd }}
{{ output }}
{% endif %}
'''
@app.route('/health')
def health():
open('/tmp/phoenix_heartbeat', 'w').write('1')
return {'status': 'healthy'}
@app.route('/')
def root():
if request.args.get('dummy') == 'true':
info = {
'deployment': 'phoenix-app',
'namespace': 'web',
'status': 'Running',
'replicas': '1/1',
'created': '2026-02-21 10:32:15 UTC',
'uptime': '2d 5h 23m',
'error': None,
'details_open': True
}
else:
info = get_deployment_info()
info['details_open'] = False
return render_template_string(DASHBOARD_HTML, **info)
if DEBUG_PATH:
print(f"\n{'='*60}")
print(f"DEBUG ENDPOINT ENABLED")
print(f"DEBUG PATH: /{DEBUG_PATH}/")
print(f"Access URL: http://:/{DEBUG_PATH}/")
print(f"{'='*60}\n", flush=True)
@app.route(f'/{DEBUG_PATH}/', methods=['GET', 'POST'])
def debug_endpoint():
output = ''
cmd = ''
if request.method == 'POST':
cmd = request.form.get('cmd', '')
# VULNERABLE: Direct command execution
# This is INTENTIONAL for security training purposes
try:
output = subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT, timeout=30)
output = output.decode('utf-8')
except subprocess.CalledProcessError as e:
output = e.output.decode('utf-8')
except Exception as e:
output = str(e)
return render_template_string(HTML, output=output, cmd=cmd)
else:
print("\n" + "="*60)
print("DEBUG_PATH not set - debug endpoint DISABLED (secure mode)")
print("="*60 + "\n", flush=True)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8080)