AIL/ail-framework
7
0
Fork 0
mirror of https://github.com/ail-project/ail-framework.git synced 2024-11-22 14:17:16 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

chg: [api] refactor blueprint

Browse source
This commit is contained in:
terrtia 2024-02-26 15:35:48 +01:00
parent 40b1378b30
commit ad63651838
No known key found for this signature in database
GPG key ID: 1E1B1F50D84613D0
7 changed files with 386 additions and 283 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

83
bin/lib/ail_api.py Executable file
View file

@ -0,0 +1,83 @@
#!/usr/bin/env python3
# -*-coding:UTF-8 -*
import os
import re
import sys
sys.path.append(os.environ['AIL_BIN'])
##################################
# Import Project packages
##################################
from lib.ConfigLoader import ConfigLoader
from lib import Users
config_loader = ConfigLoader()
r_cache = config_loader.get_redis_conn("Redis_Cache")
config_loader = None
def check_token_format(token, search=re.compile(r'[^a-zA-Z0-9_-]').search): ####################################################
return not bool(search(token))
def is_valid_token(token):
return Users.exists_token(token)
def get_user_from_token(token):
return Users.get_token_user(token)
def is_user_in_role(role, token): # verify_user_role
# User without API
if role == 'user_no_api':
return False
user_id = get_user_from_token(token)
if user_id:
return Users.is_in_role(user_id, role)
else:
return False
#### Brute Force Protection ####
def get_failed_login(ip_address):
return r_cache.get(f'failed_login_ip_api:{ip_address}')
def incr_failed_login(ip_address):
r_cache.incr(f'failed_login_ip_api:{ip_address}')
r_cache.expire(f'failed_login_ip_api:{ip_address}', 300)
def get_brute_force_ttl(ip_address):
return r_cache.ttl(f'failed_login_ip_api:{ip_address}')
def is_brute_force_protected(ip_address):
failed_login = get_failed_login(ip_address)
if failed_login:
failed_login = int(failed_login)
if failed_login >= 5:
return True
else:
return False
return False
#### --Brute Force Protection-- ####
def authenticate_user(token, ip_address):
if is_brute_force_protected(ip_address):
ip_ttl = get_brute_force_ttl(ip_address)
return {'status': 'error', 'reason': f'Max Connection Attempts reached, Please wait {ip_ttl}s'}, 401
try:
if len(token) != 55:
return {'status': 'error', 'reason': 'Invalid Token Length, required==55'}, 400
if not check_token_format(token):
return {'status': 'error', 'reason': 'Malformed Authentication String'}, 400
if is_valid_token(token):
return True, 200
# Failed Login
else:
incr_failed_login(ip_address)
return {'status': 'error', 'reason': 'Authentication failed'}, 401
except Exception as e:
print(e) # TODO Logs
return {'status': 'error', 'reason': 'Malformed Authentication String'}, 400

12
bin/lib/ail_core.py
View file

@ -3,7 +3,7 @@
import os
import sys
from uuid import uuid4
import uuid
sys.path.append(os.environ['AIL_BIN'])
##################################
@ -31,8 +31,16 @@ def _set_ail_uuid():
r_serv_db.set('ail:uuid', ail_uuid)
return ail_uuid
def is_valid_uuid_v4(header_uuid):
try:
header_uuid = header_uuid.replace('-', '')
uuid_test = uuid.UUID(hex=header_uuid, version=4)
return uuid_test.hex == header_uuid
except:
return False
def generate_uuid():
return str(uuid4())
return str(uuid.uuid4())
#### AIL OBJECTS ####

3
bin/lib/exceptions.py
View file

@ -19,3 +19,6 @@ class ModuleQueueError(AILError):
class MISPConnectionError(AILError):
pass
class AILObjectUnknown(AILError):
pass

54
bin/lib/objects/ail_objects.py
View file

@ -7,6 +7,10 @@ sys.path.append(os.environ['AIL_BIN'])
##################################
# Import Project packages
##################################
from lib.exceptions import AILObjectUnknown
from lib.ConfigLoader import ConfigLoader
from lib.ail_core import get_all_objects, get_object_all_subtypes
from lib import correlations_engine
@ -23,7 +27,7 @@ from lib.objects.Cves import Cve
from lib.objects.Decodeds import Decoded, get_all_decodeds_objects, get_nb_decodeds_objects
from lib.objects.Domains import Domain
from lib.objects import Etags
from lib.objects.Favicons import Favicon
from lib.objects import Favicons
from lib.objects import FilesNames
from lib.objects import HHHashs
from lib.objects.Items import Item, get_all_items_objects, get_nb_items_objects
@ -55,20 +59,16 @@ def sanitize_objs_types(objs):
l_types = get_all_objects()
return l_types
#### OBJECT ####
def get_object(obj_type, subtype, obj_id):
if not subtype:
if obj_type == 'item':
return Item(obj_id)
elif obj_type == 'domain':
return Domain(obj_id)
elif obj_type == 'decoded':
return Decoded(obj_id)
elif obj_type == 'chat':
return Chats.Chat(obj_id, subtype)
elif obj_type == 'chat-subchannel':
return ChatSubChannels.ChatSubChannel(obj_id, subtype)
elif obj_type == 'chat-thread':
return ChatThreads.ChatThread(obj_id, subtype)
elif obj_type == 'cookie-name':
return CookiesNames.CookieName(obj_id)
elif obj_type == 'cve':
@ -76,7 +76,7 @@ def get_object(obj_type, subtype, obj_id):
elif obj_type == 'etag':
return Etags.Etag(obj_id)
elif obj_type == 'favicon':
return Favicon(obj_id)
return Favicons.Favicon(obj_id)
elif obj_type == 'file-name':
return FilesNames.FileName(obj_id)
elif obj_type == 'hhhash':
@ -87,20 +87,42 @@ def get_object(obj_type, subtype, obj_id):
return Message(obj_id)
elif obj_type == 'screenshot':
return Screenshot(obj_id)
elif obj_type == 'title':
return Titles.Title(obj_id)
else:
raise AILObjectUnknown(f'Unknown AIL object: {obj_type} {subtype} {obj_id}')
# SUBTYPES
else:
if obj_type == 'chat':
return Chats.Chat(obj_id, subtype)
elif obj_type == 'chat-subchannel':
return ChatSubChannels.ChatSubChannel(obj_id, subtype)
elif obj_type == 'chat-thread':
return ChatThreads.ChatThread(obj_id, subtype)
elif obj_type == 'cryptocurrency':
return CryptoCurrencies.CryptoCurrency(obj_id, subtype)
elif obj_type == 'pgp':
return Pgps.Pgp(obj_id, subtype)
elif obj_type == 'title':
return Titles.Title(obj_id)
elif obj_type == 'user-account':
return UserAccount(obj_id, subtype)
elif obj_type == 'username':
return Usernames.Username(obj_id, subtype)
else:
raise Exception(f'Unknown AIL object: {obj_type} {subtype} {obj_id}')
raise AILObjectUnknown(f'Unknown AIL object: {obj_type} {subtype} {obj_id}')
def get_objects(objects):
def exists_obj(obj_type, subtype, obj_id):
obj = get_object(obj_type, subtype, obj_id)
if obj:
return obj.exists()
else:
return False
#########################################################################################
#########################################################################################
#########################################################################################
def get_objects(objects): # TODO RENAME ME
objs = set()
for obj in objects:
if isinstance(obj, dict):
@ -119,14 +141,6 @@ def get_objects(objects):
return ail_objects
def exists_obj(obj_type, subtype, obj_id):
obj = get_object(obj_type, subtype, obj_id)
if obj:
return obj.exists()
else:
return False
def get_obj_global_id(obj_type, subtype, obj_id):
obj = get_object(obj_type, subtype, obj_id)
return obj.get_global_id()

9
var/www/Flask_server.py
View file

@ -53,6 +53,8 @@ from blueprints.objects_hhhash import objects_hhhash
from blueprints.chats_explorer import chats_explorer
from blueprints.objects_image import objects_image
from blueprints.objects_favicon import objects_favicon
from blueprints.api_rest import api_rest
Flask_dir = os.environ['AIL_FLASK']
@ -113,6 +115,7 @@ app.register_blueprint(objects_hhhash, url_prefix=baseUrl)
app.register_blueprint(chats_explorer, url_prefix=baseUrl)
app.register_blueprint(objects_image, url_prefix=baseUrl)
app.register_blueprint(objects_favicon, url_prefix=baseUrl)
app.register_blueprint(api_rest, url_prefix=baseUrl)
# ========= =========#
@ -125,8 +128,6 @@ login_manager = LoginManager()
login_manager.login_view = 'root.login'
login_manager.init_app(app)
print()
# ========= LOGIN MANAGER ========
@login_manager.user_loader
@ -257,6 +258,10 @@ default_taxonomies = ["infoleak", "gdpr", "fpf", "dark-web"]
for taxonomy in default_taxonomies:
Tag.enable_taxonomy_tags(taxonomy)
# rrrr = [str(p) for p in app.url_map.iter_rules()]
# for p in rrrr:
# print(p)
# ============ MAIN ============
if __name__ == "__main__":

187
var/www/blueprints/api_rest.py Normal file
View file

@ -0,0 +1,187 @@
#!/usr/bin/env python3
# -*-coding:UTF-8 -*
"""
Blueprint Flask: crawler splash endpoints: dashboard, onion crawler ...
"""
import os
import sys
import json
from functools import wraps
from flask import request, Blueprint, Response
sys.path.append(os.environ['AIL_BIN'])
##################################
# Import Project packages
##################################
from lib import ail_api
from lib import ail_core
from lib import ail_updates
from lib import crawlers
from lib import Tag
from lib.objects import ail_objects
from importer.FeederImporter import api_add_json_feeder_to_queue
from lib.objects import Domains
from lib.objects import Titles
# ============ BLUEPRINT ============
api_rest = Blueprint('api_rest', __name__, template_folder=os.path.join(os.environ['AIL_FLASK'], 'templates'))
# ============ AUTH FUNCTIONS ============
def get_auth_from_header():
token = request.headers.get('Authorization').replace(' ', '') # remove space
return token
def token_required(user_role):
def actual_decorator(funct):
@wraps(funct)
def api_token(*args, **kwargs):
# Check AUTH Header
if not request.headers.get('Authorization'):
return create_json_response({'status': 'error', 'reason': 'Authentication needed'}, 401)
# Check Role
if not user_role:
return create_json_response({'status': 'error', 'reason': 'Invalid Role'}, 401)
token = get_auth_from_header()
ip_source = request.remote_addr
data, status_code = ail_api.authenticate_user(token, ip_address=ip_source)
if status_code != 200:
return create_json_response(data, status_code)
elif data:
# check user role
if not ail_api.is_user_in_role(user_role, token):
return create_json_response({'status': 'error', 'reason': 'Access Forbidden'}, 403)
else:
# User Authenticated + In Role
return funct(*args, **kwargs)
else:
return create_json_response({'status': 'error', 'reason': 'Internal'}, 400)
return api_token
return actual_decorator
# ============ FUNCTIONS ============
def create_json_response(data, status_code): # TODO REMOVE INDENT ????????????????????
return Response(json.dumps(data, indent=2, sort_keys=True), mimetype='application/json'), status_code
# ============= ROUTES ==============
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # CORE # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
@api_rest.route("api/v1/ping", methods=['GET'])
@token_required('read_only')
def v1_ping():
return create_json_response({'status': 'pong'}, 200)
@api_rest.route("api/v1/uuid", methods=['GET'])
@token_required('read_only')
def v1_uuid():
ail_uid = ail_core.get_ail_uuid()
return create_json_response({'uuid': ail_uid}, 200)
@api_rest.route("api/v1/version", methods=['GET'])
@token_required('read_only')
def v1_version():
version = ail_updates.get_ail_version()
return create_json_response({'version': version}, 200)
@api_rest.route("api/v1/pyail/version", methods=['GET'])
@token_required('read_only')
def v1_pyail_version():
ail_version = 'v1.0.0'
return create_json_response({'version': ail_version}, 200)
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # CRAWLERS # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # TODO: ADD RESULT JSON Response
@api_rest.route("api/v1/add/crawler/task", methods=['POST']) # TODO V2 Migration
@token_required('analyst')
def add_crawler_task():
data = request.get_json()
user_token = get_auth_from_header()
user_id = ail_api.get_user_from_token(user_token)
res = crawlers.api_add_crawler_task(data, user_id=user_id)
if res:
return create_json_response(res[0], res[1])
dict_res = {'url': data['url']}
return create_json_response(dict_res, 200)
@api_rest.route("api/v1/add/crawler/capture", methods=['POST']) # TODO V2 Migration
@token_required('analyst')
def add_crawler_capture():
data = request.get_json()
user_token = get_auth_from_header()
user_id = ail_api.get_user_from_token(user_token)
res = crawlers.api_add_crawler_capture(data, user_id)
if res:
return create_json_response(res[0], res[1])
dict_res = {'url': data['url']}
return create_json_response(dict_res, 200)
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # IMPORTERS # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
@api_rest.route("api/v1/import/json/item", methods=['POST']) # TODO V2 Migration
@token_required('user')
def import_json_item():
data_json = request.get_json()
res = api_add_json_feeder_to_queue(data_json)
return Response(json.dumps(res[0]), mimetype='application/json'), res[1]
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # OBJECTS # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # TITLES # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
@api_rest.route("api/v1/titles/download", methods=['GET'])
@token_required('analyst')
def objects_titles_download():
return create_json_response(Titles.Titles().get_contents_ids(), 200)
# TODO
@api_rest.route("api/v1/titles/download/unsafe", methods=['GET']) # TODO REFACTOR ME
@token_required('analyst')
def objects_titles_download_unsafe():
all_titles = {}
unsafe_tags = Tag.unsafe_tags
for tag in unsafe_tags:
domains = Tag.get_tag_objects(tag, 'domain')
for domain_id in domains:
domain = Domains.Domain(domain_id)
domain_titles = domain.get_correlation('title').get('title', [])
for dt in domain_titles:
title = Titles.Title(dt[1:])
title_content = title.get_content()
if title_content and title_content != 'None':
if title_content not in all_titles:
all_titles[title_content] = []
all_titles[title_content].append(domain.get_id())
return Response(json.dumps(all_titles), mimetype='application/json'), 200

259
var/www/modules/restApi/Flask_restApi.py
View file

@ -5,153 +5,39 @@
Flask functions and routes for the rest api
"""
import os
import re
import sys
import uuid
import json
# import os
# import re
# import sys
# import uuid
# import json
sys.path.append(os.environ['AIL_BIN'])
##################################
# Import Project packages
##################################
from lib.ConfigLoader import ConfigLoader
from lib import crawlers
from lib import Users
from lib.objects import Items
from lib.objects import Titles
from lib.objects import Domains
from lib import Tag
from lib import Tracker
from packages import Import_helper
from importer.FeederImporter import api_add_json_feeder_to_queue
from flask import jsonify, request, Blueprint, redirect, url_for, Response
from functools import wraps
# sys.path.append(os.environ['AIL_BIN'])
# ##################################
# # Import Project packages
# ##################################
# from lib.ConfigLoader import ConfigLoader
# from lib import Users
# from lib.objects import Items
# from lib import Tag
#
# from packages import Import_helper
#
# from importer.FeederImporter import api_add_json_feeder_to_queue
#
#
# from flask import jsonify, request, Blueprint, redirect, url_for, Response
#
# from functools import wraps
# ============ VARIABLES ============
config_loader = ConfigLoader()
baseUrl = config_loader.get_config_str("Flask", "baseurl")
baseUrl = baseUrl.replace('/', '')
if baseUrl != '':
baseUrl = '/' + baseUrl
r_cache = config_loader.get_redis_conn("Redis_Cache")
config_loader = None
import Flask_config
app = Flask_config.app
restApi = Blueprint('restApi', __name__, template_folder='templates')
# ============ AUTH FUNCTIONS ============
def check_token_format(token, search=re.compile(r'[^a-zA-Z0-9_-]').search):
return not bool(search(token))
def verify_token(token):
if len(token) != 55:
return False
if not check_token_format(token):
return False
return Users.exists_token(token)
def verify_user_role(role, token):
# User without API
if role == 'user_no_api':
return False
user_id = Users.get_token_user(token)
if user_id:
return Users.is_in_role(user_id, role)
else:
return False
# ============ DECORATOR ============
def token_required(user_role):
def actual_decorator(funct):
@wraps(funct)
def api_token(*args, **kwargs):
data = auth_errors(user_role)
if data:
return Response(json.dumps(data[0], indent=2, sort_keys=True), mimetype='application/json'), data[1]
else:
return funct(*args, **kwargs)
return api_token
return actual_decorator
def get_auth_from_header():
token = request.headers.get('Authorization').replace(' ', '') # remove space
return token
def auth_errors(user_role):
# Check auth
if not request.headers.get('Authorization'):
return {'status': 'error', 'reason': 'Authentication needed'}, 401
token = get_auth_from_header()
data = None
# verify token format
# brute force protection
current_ip = request.remote_addr
login_failed_ip = r_cache.get(f'failed_login_ip_api:{current_ip}')
# brute force by ip
if login_failed_ip:
login_failed_ip = int(login_failed_ip)
if login_failed_ip >= 5:
return {'status': 'error',
'reason': 'Max Connection Attempts reached, Please wait {}s'.format(r_cache.ttl('failed_login_ip_api:{}'.format(current_ip)))
}, 401
try:
authenticated = False
if verify_token(token): # TODO Improve Returned error
authenticated = True
# check user role
if not verify_user_role(user_role, token):
data = ({'status': 'error', 'reason': 'Access Forbidden'}, 403)
if not authenticated:
r_cache.incr(f'failed_login_ip_api:{current_ip}')
r_cache.expire(f'failed_login_ip_api:{current_ip}', 300)
data = ({'status': 'error', 'reason': 'Authentication failed'}, 401)
except Exception as e:
print(e)
data = ({'status': 'error', 'reason': 'Malformed Authentication String'}, 400)
if data:
return data
else:
return None
# ============ API CORE =============
def create_json_response(data_dict, response_code):
return Response(json.dumps(data_dict, indent=2, sort_keys=True), mimetype='application/json'), int(response_code)
def get_mandatory_fields(json_data, required_fields):
for field in required_fields:
if field not in json_data:
return {'status': 'error', 'reason': 'mandatory field: {} not provided'.format(field)}, 400
return None
# ============ FUNCTIONS ============
def is_valid_uuid_v4(header_uuid):
try:
header_uuid = header_uuid.replace('-', '')
uuid_test = uuid.UUID(hex=header_uuid, version=4)
return uuid_test.hex == header_uuid
except:
return False
# ============= ROUTES ==============
@ -160,6 +46,8 @@ def is_valid_uuid_v4(header_uuid):
# def api():
# return 'api doc'
'''
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# POST
#
@ -303,11 +191,12 @@ def get_item_content_encoded_text():
def get_item_sources():
res = Item.api_get_items_sources()
return Response(json.dumps(res[0], indent=2, sort_keys=True), mimetype='application/json'), res[1]
'''
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # TAGS # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
'''
@restApi.route("api/v1/get/tag/metadata", methods=['POST'])
@token_required('read_only')
def get_tag_metadata():
@ -323,6 +212,7 @@ def get_tag_metadata():
def get_all_tags():
res = {'tags': Tag.get_all_tags()}
return Response(json.dumps(res, indent=2, sort_keys=True), mimetype='application/json'), 200
'''
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # TODO
# # # # # # # # # # # # # # TRACKER # # # # # # # # # # # # # # # # # TODO
@ -506,42 +396,11 @@ def get_item_cryptocurrency_bitcoin():
return Response(json.dumps(res[0], indent=2, sort_keys=True), mimetype='application/json'), res[1]
'''
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # CRAWLER # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # TODO: ADD RESULT JSON Response
# @restApi.route("api/v1/crawler/task/add", methods=['POST'])
@restApi.route("api/v1/add/crawler/task", methods=['POST'])
@token_required('analyst')
def add_crawler_task():
data = request.get_json()
user_token = get_auth_from_header()
user_id = Users.get_token_user(user_token)
res = crawlers.api_add_crawler_task(data, user_id=user_id)
if res:
return create_json_response(res[0], res[1])
dict_res = {'url': data['url']}
return create_json_response(dict_res, 200)
@restApi.route("api/v1/add/crawler/capture", methods=['POST'])
@token_required('analyst')
def add_crawler_capture():
data = request.get_json()
user_token = get_auth_from_header()
user_id = Users.get_token_user(user_token)
res = crawlers.api_add_crawler_capture(data, user_id)
if res:
return create_json_response(res[0], res[1])
dict_res = {'url': data['url']}
return create_json_response(dict_res, 200)
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # DOMAIN # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
'''
@restApi.route("api/v1/get/domain/status/minimal", methods=['POST'])
@token_required('analyst')
def get_domain_status_minimal():
@ -558,6 +417,7 @@ def get_domain_status_minimal():
# res = Domain.api_get_domain_up_range(domain)
res[0]['domain'] = domain
return create_json_response(res[0], res[1])
'''
# @restApi.route("api/v1/get/crawled/domain/list", methods=['POST'])
# @token_required('analyst')
@ -601,6 +461,7 @@ def get_domain_status_minimal():
# response: {"uuid": "uuid"}
#
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
'''
@restApi.route("api/v1/import/item", methods=['POST'])
@token_required('analyst')
def import_item():
@ -664,62 +525,4 @@ def import_item_uuid():
return Response(json.dumps(data[0]), mimetype='application/json'), data[1]
return Response(json.dumps({'status': 'error', 'reason': 'Invalid response'}), mimetype='application/json'), 400
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
#
#
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
@restApi.route("api/v1/import/json/item", methods=['POST'])
@token_required('user')
def import_json_item():
data_json = request.get_json()
res = api_add_json_feeder_to_queue(data_json)
return Response(json.dumps(res[0]), mimetype='application/json'), res[1]
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # CORE # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
#
#
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
@restApi.route("api/v1/ping", methods=['GET'])
@token_required('read_only')
def v1_ping():
return Response(json.dumps({'status': 'pong'}), mimetype='application/json'), 200
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # OTHERS # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
@restApi.route("api/v1/titles/download", methods=['GET'])
@token_required('analyst')
def objects_titles_download():
return Response(json.dumps(Titles.Titles().get_contents_ids()), mimetype='application/json'), 200
@restApi.route("api/v1/titles/download/unsafe", methods=['GET'])
@token_required('analyst')
def objects_titles_download_unsafe():
all_titles = {}
unsafe_tags = Tag.unsafe_tags
for tag in unsafe_tags:
domains = Tag.get_tag_objects(tag, 'domain')
for domain_id in domains:
domain = Domains.Domain(domain_id)
domain_titles = domain.get_correlation('title').get('title', [])
for titl in domain_titles:
title = Titles.Title(titl[1:])
title_content = title.get_content()
if title_content and title_content != 'None':
if title_content not in all_titles:
all_titles[title_content] = []
all_titles[title_content].append(domain.get_id())
return Response(json.dumps(all_titles), mimetype='application/json'), 200
# ========= REGISTRATION =========
app.register_blueprint(restApi, url_prefix=baseUrl)
'''