add connections
This commit is contained in:
@@ -11,6 +11,8 @@ import jwt
|
||||
from werkzeug.security import generate_password_hash
|
||||
import secrets
|
||||
from flask_login import login_user
|
||||
import requests
|
||||
import json
|
||||
|
||||
admin_api = Blueprint('admin_api', __name__)
|
||||
|
||||
@@ -526,4 +528,334 @@ def resend_setup_mail(current_user, user_id):
|
||||
db.session.add(mail)
|
||||
|
||||
db.session.commit()
|
||||
return jsonify({'message': 'Setup mail queued for resending'})
|
||||
return jsonify({'message': 'Setup mail queued for resending'})
|
||||
|
||||
# Connection Settings
|
||||
@admin_api.route('/test-portainer-connection', methods=['POST'])
|
||||
@csrf.exempt
|
||||
def test_portainer_connection():
|
||||
data = request.get_json()
|
||||
url = data.get('url')
|
||||
api_key = data.get('api_key')
|
||||
|
||||
if not url or not api_key:
|
||||
return jsonify({'error': 'Missing required fields'}), 400
|
||||
|
||||
try:
|
||||
# Test Portainer connection
|
||||
response = requests.get(
|
||||
f"{url.rstrip('/')}/api/status",
|
||||
headers={
|
||||
'X-API-Key': api_key,
|
||||
'Accept': 'application/json'
|
||||
},
|
||||
timeout=5
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
return jsonify({'message': 'Connection successful'})
|
||||
else:
|
||||
return jsonify({'error': 'Failed to connect to Portainer'}), 400
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
@admin_api.route('/test-nginx-connection', methods=['POST'])
|
||||
@csrf.exempt
|
||||
def test_nginx_connection():
|
||||
data = request.get_json()
|
||||
url = data.get('url')
|
||||
username = data.get('username')
|
||||
password = data.get('password')
|
||||
|
||||
if not url or not username or not password:
|
||||
return jsonify({'error': 'Missing required fields'}), 400
|
||||
|
||||
try:
|
||||
# Test NGINX Proxy Manager connection
|
||||
response = requests.get(
|
||||
f"{url.rstrip('/')}/api/nginx/proxy-hosts",
|
||||
auth=(username, password),
|
||||
headers={'Accept': 'application/json'},
|
||||
timeout=5
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
return jsonify({'message': 'Connection successful'})
|
||||
else:
|
||||
return jsonify({'error': 'Failed to connect to NGINX Proxy Manager'}), 400
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
@admin_api.route('/save-portainer-connection', methods=['POST'])
|
||||
@csrf.exempt
|
||||
@token_required
|
||||
def save_portainer_connection(current_user):
|
||||
if not current_user.is_admin:
|
||||
return jsonify({'error': 'Unauthorized'}), 403
|
||||
|
||||
data = request.get_json()
|
||||
url = data.get('url')
|
||||
api_key = data.get('api_key')
|
||||
|
||||
if not url or not api_key:
|
||||
return jsonify({'error': 'Missing required fields'}), 400
|
||||
|
||||
try:
|
||||
# Save Portainer settings
|
||||
KeyValueSettings.set_value('portainer_settings', {
|
||||
'url': url,
|
||||
'api_key': api_key
|
||||
})
|
||||
|
||||
return jsonify({'message': 'Settings saved successfully'})
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
@admin_api.route('/save-nginx-connection', methods=['POST'])
|
||||
@csrf.exempt
|
||||
@token_required
|
||||
def save_nginx_connection(current_user):
|
||||
if not current_user.is_admin:
|
||||
return jsonify({'error': 'Unauthorized'}), 403
|
||||
|
||||
data = request.get_json()
|
||||
url = data.get('url')
|
||||
username = data.get('username')
|
||||
password = data.get('password')
|
||||
|
||||
if not url or not username or not password:
|
||||
return jsonify({'error': 'Missing required fields'}), 400
|
||||
|
||||
try:
|
||||
# Save NGINX Proxy Manager settings
|
||||
KeyValueSettings.set_value('nginx_settings', {
|
||||
'url': url,
|
||||
'username': username,
|
||||
'password': password
|
||||
})
|
||||
|
||||
return jsonify({'message': 'Settings saved successfully'})
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
@admin_api.route('/generate-gitea-token', methods=['POST'])
|
||||
@csrf.exempt
|
||||
def generate_gitea_token():
|
||||
"""Generate a new Gitea API token"""
|
||||
data = request.get_json()
|
||||
if not data or 'url' not in data or 'username' not in data or 'password' not in data:
|
||||
return jsonify({'message': 'Missing required fields'}), 400
|
||||
|
||||
try:
|
||||
headers = {
|
||||
'Content-Type': 'application/json'
|
||||
}
|
||||
|
||||
if data.get('otp'):
|
||||
headers['X-Gitea-OTP'] = data['otp']
|
||||
|
||||
# Generate token with required scopes
|
||||
token_data = {
|
||||
'name': f'docupulse_{datetime.utcnow().strftime("%Y%m%d_%H%M%S")}',
|
||||
'scopes': [
|
||||
'read:activitypub',
|
||||
'read:issue',
|
||||
'write:misc',
|
||||
'read:notification',
|
||||
'read:organization',
|
||||
'read:package',
|
||||
'read:repository',
|
||||
'read:user'
|
||||
]
|
||||
}
|
||||
|
||||
# Make request to Gitea API
|
||||
response = requests.post(
|
||||
f'{data["url"]}/api/v1/users/{data["username"]}/tokens',
|
||||
headers=headers,
|
||||
json=token_data,
|
||||
auth=(data['username'], data['password'])
|
||||
)
|
||||
|
||||
if response.status_code == 201:
|
||||
token_data = response.json()
|
||||
return jsonify({
|
||||
'token': token_data['sha1'],
|
||||
'name': token_data['name'],
|
||||
'token_last_eight': token_data['token_last_eight']
|
||||
}), 200
|
||||
else:
|
||||
return jsonify({'message': f'Failed to generate token: {response.json().get("message", "Unknown error")}'}), 400
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({'message': f'Failed to generate token: {str(e)}'}), 400
|
||||
|
||||
@admin_api.route('/list-gitea-repos', methods=['POST'])
|
||||
@csrf.exempt
|
||||
def list_gitea_repos():
|
||||
"""List repositories from Gitea"""
|
||||
data = request.get_json()
|
||||
if not data or 'url' not in data or 'token' not in data:
|
||||
return jsonify({'message': 'Missing required fields'}), 400
|
||||
|
||||
try:
|
||||
# Try different authentication methods
|
||||
headers = {
|
||||
'Accept': 'application/json'
|
||||
}
|
||||
|
||||
# First try token in Authorization header
|
||||
headers['Authorization'] = f'token {data["token"]}'
|
||||
|
||||
# Get user's repositories
|
||||
response = requests.get(
|
||||
f'{data["url"]}/api/v1/user/repos',
|
||||
headers=headers
|
||||
)
|
||||
|
||||
# If that fails, try token as query parameter
|
||||
if response.status_code != 200:
|
||||
response = requests.get(
|
||||
f'{data["url"]}/api/v1/user/repos?token={data["token"]}',
|
||||
headers={'Accept': 'application/json'}
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
return jsonify({
|
||||
'repositories': response.json()
|
||||
}), 200
|
||||
else:
|
||||
return jsonify({'message': f'Failed to list repositories: {response.json().get("message", "Unknown error")}'}), 400
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({'message': f'Failed to list repositories: {str(e)}'}), 400
|
||||
|
||||
@admin_api.route('/list-gitlab-repos', methods=['POST'])
|
||||
@csrf.exempt
|
||||
def list_gitlab_repos():
|
||||
"""List repositories from GitLab"""
|
||||
data = request.get_json()
|
||||
if not data or 'url' not in data or 'token' not in data:
|
||||
return jsonify({'message': 'Missing required fields'}), 400
|
||||
|
||||
try:
|
||||
headers = {
|
||||
'PRIVATE-TOKEN': data['token'],
|
||||
'Accept': 'application/json'
|
||||
}
|
||||
|
||||
# Get user's projects (repositories)
|
||||
response = requests.get(
|
||||
f'{data["url"]}/api/v4/projects',
|
||||
headers=headers,
|
||||
params={'membership': 'true'} # Only get projects where user is a member
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
return jsonify({
|
||||
'repositories': response.json()
|
||||
}), 200
|
||||
else:
|
||||
return jsonify({'message': f'Failed to list repositories: {response.json().get("message", "Unknown error")}'}), 400
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({'message': f'Failed to list repositories: {str(e)}'}), 400
|
||||
|
||||
@admin_api.route('/test-git-connection', methods=['POST'])
|
||||
@csrf.exempt
|
||||
def test_git_connection():
|
||||
"""Test the connection to a Git repository"""
|
||||
data = request.get_json()
|
||||
if not data or 'provider' not in data or 'url' not in data or 'username' not in data or 'token' not in data or 'repo' not in data:
|
||||
return jsonify({'message': 'Missing required fields'}), 400
|
||||
|
||||
try:
|
||||
if data['provider'] == 'gitea':
|
||||
# Test Gitea connection with different authentication methods
|
||||
headers = {
|
||||
'Accept': 'application/json'
|
||||
}
|
||||
|
||||
# First try token in Authorization header
|
||||
headers['Authorization'] = f'token {data["token"]}'
|
||||
|
||||
# Try to get repository information
|
||||
response = requests.get(
|
||||
f'{data["url"]}/api/v1/repos/{data["repo"]}',
|
||||
headers=headers
|
||||
)
|
||||
|
||||
# If that fails, try token as query parameter
|
||||
if response.status_code != 200:
|
||||
response = requests.get(
|
||||
f'{data["url"]}/api/v1/repos/{data["repo"]}?token={data["token"]}',
|
||||
headers={'Accept': 'application/json'}
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
return jsonify({'message': 'Connection successful'}), 200
|
||||
else:
|
||||
return jsonify({'message': f'Connection failed: {response.json().get("message", "Unknown error")}'}), 400
|
||||
|
||||
elif data['provider'] == 'gitlab':
|
||||
# Test GitLab connection
|
||||
headers = {
|
||||
'PRIVATE-TOKEN': data['token'],
|
||||
'Accept': 'application/json'
|
||||
}
|
||||
|
||||
# Try to get repository information
|
||||
response = requests.get(
|
||||
f'{data["url"]}/api/v4/projects/{data["repo"].replace("/", "%2F")}',
|
||||
headers=headers
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
return jsonify({'message': 'Connection successful'}), 200
|
||||
else:
|
||||
return jsonify({'message': f'Connection failed: {response.json().get("message", "Unknown error")}'}), 400
|
||||
|
||||
else:
|
||||
return jsonify({'message': 'Invalid Git provider'}), 400
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({'message': f'Connection failed: {str(e)}'}), 400
|
||||
|
||||
@admin_api.route('/save-git-connection', methods=['POST'])
|
||||
@csrf.exempt
|
||||
@token_required
|
||||
def save_git_connection(current_user):
|
||||
if not current_user.is_admin:
|
||||
return jsonify({'error': 'Unauthorized'}), 403
|
||||
|
||||
data = request.get_json()
|
||||
provider = data.get('provider')
|
||||
url = data.get('url')
|
||||
username = data.get('username')
|
||||
token = data.get('token')
|
||||
repo = data.get('repo')
|
||||
|
||||
if not provider or not url or not username or not token or not repo:
|
||||
return jsonify({'error': 'Missing required fields'}), 400
|
||||
|
||||
if provider not in ['gitea', 'gitlab']:
|
||||
return jsonify({'error': 'Invalid provider'}), 400
|
||||
|
||||
try:
|
||||
# Save Git settings
|
||||
KeyValueSettings.set_value('git_settings', {
|
||||
'provider': provider,
|
||||
'url': url,
|
||||
'username': username,
|
||||
'token': token,
|
||||
'repo': repo
|
||||
})
|
||||
|
||||
return jsonify({'message': 'Settings saved successfully'})
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
Reference in New Issue
Block a user