paramecio/pastafari2
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity

Added api to pastafari

Browse source
This commit is contained in:
Antonio de la Rosa 2025-04-28 00:54:41 +02:00
parent 96316454c2
commit ec52e45969
3 changed files with 608 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

39
__apidoc__.py Normal file
View file

@ -0,0 +1,39 @@
from modules.apidoc2.libraries.apidoc import AppDoc
from modules.pastafari2 import pastafari_app
from settings import config
description="""
An API for use with Pastafari2 paramecio module.
"""
appdoc=AppDoc(pastafari_app, description)
appdoc.url='/'
appdoc.routes['pastafari_app.get_groups'].summary='Get category groups of servers'
appdoc.routes['pastafari_app.get_groups'].description='List with the server groups'
#appdoc.routes['pastafari_app.get_groups'].add_parameter(config.cookie_name, 'cookie', 'Cookie for identify in api', 'true', 'string')
appdoc.routes['pastafari_app.add_group'].summary='Add group to database'
appdoc.routes['pastafari_app.add_group'].description='Add group to database. You can organize your servers on groups.'
#appdoc.routes['pastafari_app.add_group'].add_parameter(config.cookie_name, 'cookie', 'Cookie for identify in api', 'true', 'string')
appdoc.routes['pastafari_app.add_server'].summary='Add server to Pastafaridb'
appdoc.routes['pastafari_app.add_server'].description='Add server to Pastafari, for management the server using the control panel via SSH'
#appdoc.routes['pastafari_app.add_server'].add_parameter(config.cookie_name, 'cookie', 'Cookie for identify in api', 'true', 'string')
appdoc.routes['pastafari_app.make_task'].summary='Make task in server'
appdoc.routes['pastafari_app.make_task'].description='Make task in server async. You send the task to cuchulu agent, ant this execute the task in the server, and save results in log'
#appdoc.routes['pastafari_app.make_task'].add_parameter(config.cookie_name, 'cookie', 'Cookie for identify in api', 'true', 'string')
appdoc.routes['pastafari_app.add_task_db'].summary='Add task to db'
appdoc.routes['pastafari_app.add_task_db'].description='Add task to db for use in API'
#appdoc.routes['pastafari_app.add_task_db'].add_parameter(config.cookie_name, 'cookie', 'Cookie for identify in api', 'true', 'string')
appdoc.routes['pastafari_app.get_progress_task'].summary='Get progress from task'
appdoc.routes['pastafari_app.get_progress_task'].description='Get progress from task from db in logtask table'
appdoc.routes['pastafari_app.get_progress_task'].add_parameter('server', 'query', 'IP server from log', 'false', 'string')
#appdoc.routes['pastafari_app.get_progress_task'].add_parameter(config.cookie_name, 'cookie', 'Cookie for identify in api', 'true', 'string')
appdoc.tags['servers']='Servers'

503
api.py Normal file
View file

@ -0,0 +1,503 @@
from modules.pastafari2 import pastafari_app
from paramecio2.modules.admin.models.admin import UserAdmin
from paramecio2.libraries.mtemplates import PTemplate, env_theme
from paramecio2.libraries.i18n import I18n, PGetText
from paramecio2.libraries.db import corefields
from typing import Annotated
from modules.pastafari2.models.tasks import Task, LogTask, TaskDb as TaskDbModel
from modules.pastafari2.libraries.configtask import config_task
from modules.pastafari2.models.pastafari2 import ServerGroup, ServerDbTask as ServerModel, UpdateServerScripts
from settings import config
#from modules.cuchulucp.libraries.tokenplugin import token_plugin
#from bottle import request, response
from paramecio2.libraries.responsesapi import ResponseItems, ListItem, StandardResponse
from paramecio2.libraries.db.simplequery import insert, select
from modules.pastafari2.libraries.task import Task as SSHTask
from paramecio2.libraries.db.extrafields.ipfield import IpField
from paramecio2.libraries.db.extrafields.arrayfield import ArrayField
from paramecio2.libraries.db.extrafields.parentfield import ParentField
from paramecio2.libraries.db.extrafields.datefield import DateField
from paramecio2.libraries.db.extrafields.dictfield import DictField
from paramecio2.libraries.db.extrafields.jsonfield import JsonValueField
from paramecio2.libraries.formsutils import check_form
from modules.pastafari2.libraries.authplugin import auth_plugin
from paramecio2.modules.admin.libraries.check_login_tries import check_login_tries
from paramecio2.libraries.datetime import now, format_local_strtime, timestamp_to_datetime, obtain_timestamp, TimeClass
from importlib import import_module, reload
#from paramecio2.libraries.sessionplugin import SessionPlugin, get_session
from paramecio2.libraries.keyutils import create_key_encrypt
from modules.pastafari2 import pastafari_app
from paramecio2.libraries.plugins import db
from flask import g
import socket
try:
import ujson as json
except:
import json
env=env_theme(__file__)
t=PTemplate(env)
gtext=PGetText(__file__)
_=gtext.gettext
api_key=''
if hasattr(config, 'cuchulu_api_key'):
api_key=config.cuchulu_api_key
if hasattr(config, 'cuchulu_ssh_private_key'):
config_task.ssh_private_key=config.cuchulu_ssh_private_key
if hasattr(config, 'cuchulu_ssh_private_key_password'):
config_task.ssh_private_key_password=config.cuchulu_ssh_private_key_password
if hasattr(config, 'cuchulu_ssh_public_key'):
config_task.ssh_public_key=config.cuchulu_ssh_public_key
class ServerGroups(ListItem):
group=corefields.CharField('group')
code_group=corefields.CharField('code_group')
class ResponseGroups(ResponseItems):
groups=ServerGroups()
class Server:
server_host=corefields.CharField('server_host')
server_username=corefields.CharField('server_username')
server_password=corefields.CharField('server_password')
group_id=corefields.IntegerField('group_id')
class UserAdminFields:
username=corefields.CharField('username')
password=corefields.CharField('password')
class TaskLogs(ListItem):
task_id=corefields.IntegerField('task_id')
server=IpField('server')
progress=corefields.DoubleField('progress')
no_progress=corefields.BooleanField('no_progress')
message=corefields.LongTextField('message')
error=corefields.BooleanField('error')
status=corefields.BooleanField('status')
code_error=corefields.BooleanField('code_error')
class ResponseLog(ResponseItems):
log=TaskLogs()
class TaskExec:
taskdb_id=corefields.IntegerField('taskdb_id')
server_id=corefields.IntegerField('server_id')
data=DictField('data', corefields.CharField('data'))
class TaskDb:
name=corefields.CharField('name')
path=corefields.CharField('path', 4096)
class TokenResponse(StandardResponse):
token=corefields.CharField('token')
groupdb=ServerGroup(None)
@pastafari_app.get('/api/v1/get_groups')
@db
def get_groups(tag='groups') -> ResponseGroups:
db=g.connection
arr_group=select(groupdb, db)
resp=ResponseGroups()
resp.groups=arr_group
return resp.toDict()
@pastafari_app.post('/api/v1/add_group')
@db
def add_group(post: Annotated[ServerGroups, 'Add group to the cuchulucp system'] = ServerGroups, tag='groups'):
db=g.connection
error=True
message=_('Cannot insert the new group')
code_error=1
json_post=request.get_json()
if json_post:
post=json_post
if insert(groupdb, db, post):
code_error=0
error=False
message=_('Success')
return {'error': error, 'message': message, 'code_error': code_error}
@pastafari_app.post('/api/v1/add_server')
@db
def add_server(post: Annotated[Server, 'Add server to the cuchulucp system'] = Server, tag='servers'):
db=g.connection
error_form={}
json_post=request.get_json()
if json_post:
server_host=json_post.get('server_host', '')
server_username=json_post.get('server_username', '')
server_password=json_post.get('server_password', '')
group_id=json_post.get('group_id', '')
private_key=config_task.ssh_private_key
password_key=config_task.ssh_private_key_password
public_key=config_task.ssh_public_key
remote_path=config_task.remote_path
task_id=0
ip=''
error=0
data={}
ssh_user=config_task.remote_user
ssh_port=config_task.port
try:
tmp_port=int(json_post.get('ssh_port', '22'))
ssh_port=tmp_port
except:
pass
ssh_port='22'
#make ping to server
if server_host=='':
error=1
error_form['#server_host_error']=_('Error: you need enter a valid hostname')
if server_username=='':
error=1
error_form['#server_username_error']=_('Error: you need enter a valid username for the server')
txt_error=''
try:
ip=socket.gethostbyname(server_host)
pass
except socket.error as err_msg:
error=1
error_form['#server_host_error']=_('Error: ')+err_msg.__str__()
if not error:
group_name=''
if group_id!='':
server_group=ServerGroup(db)
arr_group=server_group.set_conditions('WHERE id=%s', [group_id]).select_a_row_where()
group_name=arr_group['code_group']
data={'ssh_user': ssh_user, 'pub_key': public_key, 'hostname': server_host, 'ip': ip, 'group_id': group_id, 'group_name': group_name}
with SSHTask(server_host, db, remote_user=server_username, remote_password=server_password, private_key=private_key, password_key=password_key, remote_path=remote_path, task_id=task_id, data=data, port=ssh_port) as ssh_task:
if not ssh_task.prepare_connection():
error=1
txt_error=ssh_task.txt_error #_('Error: cannot connect to server')
error_form['#server_host_error']=txt_error #_('Error: cannot connect to server')
else:
# Prepare task for install monit
task=Task(db)
task_id=0
path_task='modules.cuchulucp.tasks.system.task'
if not task.run_task(ip, path_task, 'Add new server', 'add_new_server', 'Task for add a new server', user=server_username, password=server_password, where_sql_server='', url='', data=data, send_task=True, ssh_port=ssh_port):
error=1
error_form['#server_host_error']=_('Error: cannot execute the task ')+task.txt_error
txt_error=_('Error: cannot execute the task ')+task.txt_error
task_id=task.task_id
else:
error=1
txt_error=_('Error: json malformed')
error_form={}
task_id=0
return {'error': error, 'message': txt_error, 'error_form': error_form, 'task_id': task_id}
# Make task in server.
serverdb=ServerModel()
taskdb=TaskDbModel(None)
@pastafari_app.post('/api/v1/task')
@db
def make_task(post: Annotated[TaskExec, 'Task to execute'] = TaskExec, tag='tasks'):
db=g.connection
error=1
txt_error=_('Cannot execute the task')
error_form={}
task_id=0
json_post=request.get_json()
if json_post:
server_id=json_post.get('server_id', '0')
taskdb_id=json_post.get('taskdb_id', '0')
data=json_post.get('data', {})
if type(data) is not dict:
data={}
arr_server=select(serverdb, db, dict_fields=[], where_sql='WHERE id=%s', limit='', dict_values=[server_id])
if arr_server:
arr_task=select(taskdb, db, dict_fields=[], where_sql='where id=%s', limit='', dict_values=[taskdb_id])
task_file=arr_task[0]['path']
path_task=task_file
task_execute=import_module(task_file)
if config.reloader:
reload(task_execute)
user=config_task.remote_user
task_first=task_execute.ServerTask('', db, remote_user=user, remote_password='', private_key=config_task.ssh_private_key, password_key='', remote_path='cuchulucp', task_id=0, data=data)
server_host=arr_server[0]['ip']
ssh_key_priv=config_task.ssh_private_key
ssh_key_password=config_task.ssh_private_key_password
where_sql=''
task=Task(db)
# Check post
check_form_done=True
if hasattr(task_first, 'check_form'):
check_form_done=task_first.check_form(data)
if check_form_done:
if not task.run_task(server_host, path_task, task_first.name_task, task_first.codename_task, task_first.description_task, user=user, password='', where_sql_server=where_sql, ssh_key_priv=ssh_key_priv, ssh_key_password=ssh_key_password, url=task_first.url_return, data=data, send_task=True):
error=1
error_form['#server_host_error']=_('Error: cannot execute the task')
txt_error=task.txt_error
else:
error=0
txt_error=_('Success')
task_id=task.task_id
else:
error=1
for k, form in task_first.arr_form.items():
if form.error:
error_form[k]=form.txt_error
txt_error=task.txt_error
else:
txt_error=_('Error: server not exists')
else:
txt_error=_('Error: json malformed')
return {'error': error, 'message': txt_error, 'error_form': error_form, 'task_id': task_id}
@pastafari_app.get('/api/v1/get_progress_task/<int:task_id>/<int:position>')
@db
def get_progress_task(task_id, position=0, tag='tasks') -> ResponseLog:
db=g.connection
task=Task(db)
logtask=LogTask(db)
arr_task=task.select_a_row(task_id)
arr_rows={'wait': 1}
if arr_task:
logtask.set_limit([position, 50])
logtask.set_order({'id': 0})
logtask.conditions=['WHERE task_id=%s', [task_id]]
server=request.query.get('server', '')
if server!='':
logtask.conditions=['WHERE task_id=%s', [task_id]]
logtask.yes_reset_conditions=False
c=0
arr_rows=[]
with logtask.select([], True) as cursor:
for arr_row in cursor:
arr_rows.append(arr_row)
c+=1
if c==0:
arr_rows=[]
c_error=logtask.set_conditions('WHERE task_id=%s and logtask.error=%s and logtask.status=%s', [task_id, 1, 1]).select_count()
if c_error==0:
arr_rows={'wait': 1}
else:
logtask.limit=''
with logtask.set_conditions('WHERE task_id=%s and logtask.error=%s and logtask.status=%s', [task_id, 1, 1]).select([], True) as cursor:
for arr_row in cursor:
arr_rows.append(arr_row)
else:
arr_rows=[{'task_id': task_id, 'progress': 100, 'message': 'Error: no exists task', 'error': 1, 'status': 1}]
return { "code_error": 1, "error": 0, "log": arr_rows, "message": _('Successfully')}
# Add task to db for use with api, for not send python paths directly.
@pastafari_app.post('/api/v1/add_task_db')
@db
def add_task_db(post: Annotated[TaskDb, 'Add a new task to db'] = TaskDb, tag='tasks') -> StandardResponse:
db=g.connection
taskdb.conn(db)
taskdb.create_forms()
error=1
txt_error='Cannot insert the task'
error_form={}
taskdb_id=0
post=request.get_json()
if post:
if taskdb.insert(post):
error=0
txt_error=_('Success')
taskdb_id=taskdb.insert_id()
else:
error_form=taskdb.collect_errors()
else:
txt_error=_('Error: json malformed')
return {'error': error, 'message': txt_error, 'error_form': error_form, 'taskdb_id': taskdb_id}
@pastafari_app.post('/api/v1/login')
@db
def check_login_admin(post: Annotated[UserAdminFields, 'Login']=UserAdminFields, tag='login') -> TokenResponse:
db=g.connection
usermodel=UserAdmin(db)
error=1
message=_('Invalid user and password')
no_login=check_login_tries(request, db)
code_error=0
error_form={}
error_form['no_login']=no_login
json_post=request.get_json()
if json_post:
username=json_post.get('username')
password=json_post.get('password')
token_auth=''
if username!='' and password!='' and not no_login:
with db.query('select * from useradmin WHERE username=%s', [username]) as cursor:
result=cursor.fetchone()
if result:
if usermodel.fields['password'].verify(password, result['password']):
now_str=now()
date_now=format_local_strtime('YYYY-MM-DD HH:mm:ss', now_str)
with db.query('select * from usertoken WHERE user_id=%s', [result['id']]) as cursor:
arr_token=cursor.fetchone()
if not arr_token:
#print('insert token')
token_auth=create_key_encrypt(48)
if db.query('insert into usertoken (user_id, token, last_login) VALUES (%s, %s, %s)', [result['id'], token_auth, date_now]):
error=0
message=_('Success')
else:
token_auth=arr_token['token']
error=0
message=_('Success')
else:
message=_('Error: json malformed')
return {'error': error, 'message': message, 'code_error': code_error, 'error_form': error_form, 'token': token_auth}

66
libraries/authplugin.py Normal file
View file

@ -0,0 +1,66 @@
"""
Paramecio2fm is a series of wrappers for Flask, mako and others and construct a simple headless cms.
Copyright (C) 2023 Antonio de la Rosa Caballero
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
from flask import g, session, redirect, url_for
from functools import wraps
from paramecio2.libraries.db.webmodel import WebModel
def auth_plugin(f):
@wraps(f)
def wrapper(*args, **kwds):
"""Wrapper function for check login in your pastafari2 api
Wrapper function for check login in your pastafari2 api
Args:
*args (mixed): The args of function
**kwds (mixed): Standard python extra arguments of function
Returns:
wrapper (function): Return the wrapper.
"""
auth=request.headers.get('Authorization', None)
if auth:
#print(request.headers['Authorization'])
bearer=request.headers['Authorization'].replace('Bearer', '').strip()
db=kwargs['db']
num_token=0
with db.query('select count(*) as num_token from usertoken WHERE token=%s', [bearer]) as cursor:
num_token=cursor.fetchone()['num_token']
if num_token==0:
return abort(403, 'You need a valid bearer token for access')
else:
return abort(403, 'You need a valid bearer token for access')
code=f(*args, **kwds)
return code
return wrapper