Re-implement spec_version auto filtering

cti-taxii-server

This is an OASIS TC Open Repository. See the Governance section for more information.

Trusted Automated Exchange of Intelligence Information (TAXII™) is an application layer protocol for the communication of cyber threat information in a simple and scalable manner.

Medallion is a minimal implementation of a TAXII 2.1 Server in Python.

WARNING: medallion was designed as a prototype and reference implementation of TAXII 2.1, and is not intended for production use.

medallion has been designed to be a simple front-end REST server providing access to the endpoints defined in that specification. It uses the python framework - Flask. medallion depends on back-end "plugins" which handle the persistence of the TAXII data and metadata. The TAXII specification is agnostic to what type of data a TAXII server stores, but this will usually be STIX 2 content.

Two back-end plugins are provided with medallion: the Memory back-end and the MongoDB back-end. The Memory back-end persists data "in memory". It is initialized using a json file that contains TAXII data and metadata. It is possible to save the current state of the in memory store, but this back-end is really intended only for testing purposes. The MongoDB backend is somewhat more robust and makes use of a MongoDB server, installed independently. The MongoDB back-end can only be used if the pymongo python package is installed. An error message will result if it is used without that package.

For more information, see the documentation on ReadTheDocs.

Installation

The easiest way to install medallion is with pip

$ pip install medallion

Usage

As a script

Medallion provides a command-line interface to start the TAXII Server

usage: medallion [-h] [--host HOST] [--port PORT] [--debug-mode]
                 [--log-level {DEBUG,INFO,WARN,ERROR,CRITICAL}]
                 [-c CONF_FILE] [--conf-dir CONF_DIR | --no-conf-dir]
                 [--conf-check] [CONFIG_PATH]

medallion v3.0.0

positional arguments:
  CONFIG_PATH           Deprecated argument for specifying a single
                        JSON configuration file. Do not specify this
                        and `--conf-file`.

optional arguments:
  -h, --help            show this help message and exit

  --host HOST           The host to listen on.

  --port PORT           The port of the web server.

  --debug-mode          If set, start application in debug mode.

  --log-level {DEBUG,INFO,WARN,ERROR,CRITICAL}
                        The logging output level for medallion.

  -c CONF_FILE, --conf-file CONF_FILE
                        Path to a single configuration file. Defaults to the
                        value of the MEDALLION_CONFFILE environment variable
                        or /etc/xdg/medallion/3/medallion.conf.

  --conf-dir CONF_DIR   Path to a directory containing JSON configuration
                        files with names ending in .json or .conf. Defaults to
                        the value of the MEDALLION_CONFDIR environment
                        variable or /etc/xdg/medallion/3/config.d.

  --no-conf-dir         Disable the use of any configuration directory as
                        described for --conf-dir. This may be used to ensure
                        that the default or some value from the environment is
                        not used.

  --conf-check          Evaluate medallion configuration without running the
                        server.

To run medallion

$ python medallion/scripts/run.py --conf-file <config-file>

Make sure medallion is using the same port that your TAXII client will be connecting on. You can specify which port medallion runs on using the --port option, for example

$ medallion --port 80 --conf-file config_file.json

The <config_file> contains:

configuration information for the backend plugin
a simple user name/password dictionary

To use the Memory back-end plug, include the following in the <config-file>:

{
    "backend": {
        "module_class": "MemoryBackend",
        "filename": "<path to json file with initial data>"
    }
}

To use the Mongo DB back-end plug, include the following in the <config-file>:

{
     "backend": {
        "module_class": "MongoBackend",
        "uri": "<Mongo DB server url>  # e.g., 'mongodb://localhost:27017/'"
     }
}

Note: A Mongo DB should be available at some URL when using the Mongo DB back-end

A description of the Mongo DB structure expected by the mongo db backend code is described in the documentation.

As required by the TAXII specification, medallion supports HTTP Basic authorization. However, the user names and passwords are currently stored in the <config_file> in plain text.

Here is an example:

{
    "users": {
       "admin": "Password0",
       "user1": "Password1",
       "user2": "Password2"
    }
}

The authorization is enabled using the python package flask_httpauth. Authorization could be enhanced by changing the method "decorated" using @auth.get_password in medallion/__init__.py

Configs may also contain a "taxii" section as well, as shown below:

{
    "taxii": {
       "max_page_size": 100
       "interop_requirements": true
    }
}

All TAXII servers require a config, though if any of the sections specified above are missing, they will be filled with default values.

The interop_requirements option will enforce additional requireemnts from the TAXII 2.1 Interoperability specification. It defaults to false.

We welcome contributions for other back-end plugins.

Docker

We also provide a Docker image to make it easier to run medallion

$ docker build . -t medallion -f docker_utils/Dockerfile

If operating behind a proxy, add the following option (replacing <proxy> with your proxy location and port): --build-arg https_proxy=<proxy>.

Then run the image

$ docker run --rm -p 5000:5000 -v <directory>:/var/taxii medallion

Replace <directory> with the full path to the directory containing your medallion configuration.

Governance

This GitHub public repository ( https://github.com/oasis-open/cti-taxii-server ) was created at the request of the OASIS Cyber Threat Intelligence (CTI) TC as an OASIS TC Open Repository to support development of open source resources related to Technical Committee work.

While this TC Open Repository remains associated with the sponsor TC, its development priorities, leadership, intellectual property terms, participation rules, and other matters of governance are separate and distinct from the OASIS TC Process and related policies.

All contributions made to this TC Open Repository are subject to open source license terms expressed in the BSD-3-Clause License. That license was selected as the declared "Applicable License" when the TC Open Repository was created.

As documented in "Public Participation Invited", contributions to this OASIS TC Open Repository are invited from all parties, whether affiliated with OASIS or not. Participants must have a GitHub account, but no fees or OASIS membership obligations are required. Participation is expected to be consistent with the OASIS TC Open Repository Guidelines and Procedures, the open source LICENSE designated for this particular repository, and the requirement for an Individual Contributor License Agreement that governs intellectual property.

Maintainers

TC Open Repository Maintainers are responsible for oversight of this project's community development activities, including evaluation of GitHub pull requests and preserving open source principles of openness and fairness. Maintainers are recognized and trusted experts who serve to implement community goals and consensus design preferences.

Initially, the associated TC members have designated one or more persons to serve as Maintainer(s); subsequently, participating community members may select additional or substitute Maintainers, per consensus agreements.

Current Maintainers of this TC Open Repository

Chris Lenk; GitHub ID: https://github.com/clenk/; WWW: MITRE Corporation
Rich Piazza; GitHub ID: https://github.com/rpiazza/; WWW: MITRE Corporation
Zach Rush; GitHub ID: https://github.com/zrush-mitre/; WWW: MITRE Corporation
Jason Keirstead; GitHub ID: https://github.com/JasonKeirstead; WWW: IBM

About OASIS TC Open Repositories

Feedback

Questions or comments about this TC Open Repository's activities should be composed as GitHub issues or comments. If use of an issue/comment is not possible or appropriate, questions may be directed by email to the Maintainer(s) listed above. Please send general questions about Open Repository participation to OASIS Staff at [email protected] and any specific CLA-related questions to [email protected].

	with tempfile.NamedTemporaryFile() as f:
	self.client.post(
	test.ADD_OBJECTS_EP,
	data=json.dumps(new_bundle),
	headers=post_header,
	)
	self.app.medallion_backend.save_data_to_file(f.name)
	assert os.path.isfile(f.name)

	def filter_contains_marking_definition(self, pipeline):
	# If we are matching on id (either match[id]= or /{id}), then check if
	# we are trying to find a marking definition. If so, we don't want do
	# filter by version as marking-definition objects are not versioned.
	if "id" in pipeline[0]["$match"].keys() and pipeline[0]["$match"]["id"].startswith("marking-definition"):
	return True

	if "_type" in pipeline[0]["$match"].keys():
	if ((
	isinstance(pipeline[0]["$match"]["_type"], dict) and
	"$in" in pipeline[0]["$match"]["_type"].keys()
	) and
	("marking-definition" in pipeline[0]["$match"]["_type"]["$in"])):
	return True
	elif pipeline[0]["$match"]["_type"].startswith("marking-definition"):
	return True

	return False

	# need to handle marking-definitions differently as they are not versioned like SDO's
	if self.filter_contains_marking_definition(pipeline):
	# If we are finding marking-definitions from the objects collection
	# we need to change the match criteria from "_type" to "type"
	if data.name == "objects" and "_type" in pipeline[0]["$match"].keys():
	pipeline[0]["$match"]["type"] = pipeline[0]["$match"].pop("_type")

	# Calculate total number of matching documents
	if data.name == "objects":
	count = self.get_result_count(pipeline, manifest_info["mongodb_collection"])
	else:
	count = self.get_result_count(pipeline, data)

	self.add_pagination_operations(pipeline)

	cursor = data.aggregate(pipeline)
	results = list(cursor)

	return count, results

	for new_obj in objs["objects"]:
	id_and_version_already_present = False
	for obj in collection["objects"]:
	id_and_version_already_present = False

	if new_obj["id"] == obj["id"]:
	if "modified" in new_obj:
	if new_obj["modified"] == obj["modified"]:
	id_and_version_already_present = True
	else:
	# There is no modified field, so this object is immutable
	id_and_version_already_present = True
	if not id_and_version_already_present:
	collection["objects"].append(new_obj)
	self._update_manifest(new_obj, api_root, collection["id"], request_time)
	successes.append(new_obj["id"])
	succeeded += 1
	else:
	failures.append({"id": new_obj["id"], "message": "Unable to process object"})
	failed += 1

	import datetime
	import json
	import os
	import uuid

	from medallion.backends.taxii.base import Backend
	from medallion.exceptions import ProcessingError
	from medallion.filters.basic_filter import BasicFilter
	from medallion.utils.common import create_bundle, generate_status


	class DirectoryBackend(Backend):
	# access control is handled at the views level

	def __init__(self, path=None, **kwargs):
	self.path = path
	self.discovery_config = self.init_discovery_config(kwargs.get('discovery', None))
	self.api_root_config = self.init_api_root_config(kwargs.get('api-root', None))
	self.collection_config = self.init_collection_config(kwargs.get('collection', None))
	self.cache = {}
	self.statuses = []

	# noinspection PyMethodMayBeStatic
	def init_discovery_config(self, discovery_config):
	if not self.path:
	raise ProcessingError('path was not specified in the config file', 400)

	if not os.path.isdir(self.path):
	raise ProcessingError("directory '{}' was not found".format(self.path), 500)

	return discovery_config

	def update_discovery_config(self):
	dc = self.discovery_config
	collection_dirs = sorted([f for f in os.listdir(self.path) if os.path.isdir(os.path.join(self.path, f))])

	if not collection_dirs:
	raise ProcessingError('at least one api-root directory is required', 500)

	updated_roots = ['{}{}/'.format(dc['host'], f) for f in collection_dirs]

	self.discovery_config['default'] = updated_roots[0]
	self.discovery_config['api_roots'] = updated_roots

	# noinspection PyMethodMayBeStatic
	def init_api_root_config(self, api_root_config):
	if api_root_config:
	return api_root_config
	else:
	raise ProcessingError('api-root was not specified in the config file', 400)

	# noinspection PyMethodMayBeStatic
	def init_collection_config(self, collection_config):
	if collection_config:
	return collection_config
	else:
	raise ProcessingError('collection was not specified in the config file', 400)

	def validate_requested_api_root(self, requested_api_root):
	api_roots = self.discovery_config['api_roots']

	host_port = self.discovery_config['default'].rsplit('/', 2)[0]

	full_api_root = '{}/{}/'.format(host_port, requested_api_root)

	return full_api_root in api_roots

	def server_discovery(self):
	self.update_discovery_config()
	return self.discovery_config

	def get_api_root_information(self, api_root):
	self.update_discovery_config()

	api_roots = self.discovery_config['api_roots']

	for r in api_roots:
	c_dir = r.rsplit('/', 2)[1]

	if api_root == c_dir:
	i_title = "Indicators from directory '{}'".format(c_dir)

	i = {
	"title": i_title,
	"description": "",
	"versions": self.api_root_config['versions'],
	"max-content-length": self.api_root_config['max-content-length']
	}

	return i

	def get_collections(self, api_root, start_index, end_index):
	self.update_discovery_config()

	api_roots = self.discovery_config['api_roots']

	collections = []

	# Generate a collection object for each api_root
	for r in api_roots:
	c_dir = r.rsplit('/', 2)[1]

	if api_root == c_dir:
	c_id = uuid.uuid3(uuid.NAMESPACE_URL, r)
	c_title = "Indicators from directory '{}'".format(c_dir)

	c = {
	"id": str(c_id),
	"title": c_title,
	"description": self.collection_config['description'],
	"can_read": self.collection_config['can_read'],
	"can_write": self.collection_config['can_write'],
	"media_types": self.collection_config['media_types']
	}

	collections.append(c)

	count = len(collections)

	collections = collections if end_index == -1 else collections[start_index:end_index]

	return count, collections

	def get_collection(self, api_root, collection_id):
	count, collections = self.get_collections(api_root, 0, -1)

	for c in collections:
	if 'id' in c and collection_id == c['id']:
	return c

	def set_modified_time_stamp(self, objects, modified):
	for o in objects:
	o['modified'] = modified

	return objects

	def get_modified_time_stamp(self, fp):
	fp_modified = os.path.getmtime(fp)
	dt = datetime.datetime.utcfromtimestamp(fp_modified)
	modified = '{:%Y-%m-%dT%H:%M:%S.%fZ}'.format(dt)

	return modified

	def delete_from_cache(self, api_root):
	p = os.path.join(self.path, api_root)
	files = [f for f in os.listdir(p) if os.path.isfile(os.path.join(p, f)) and f.endswith('.json')]

	for f in self.cache[api_root]['files'].keys():
	if f not in files:
	del self.cache[api_root]['files'][f]

	def add_to_cache(self, api_root, api_root_modified, file_name, file_modified):
	fp = os.path.join(self.path, api_root, file_name)

	u_objects = []

	with open(fp, 'r') as raw_json:
	try:
	stix2 = json.load(raw_json)

	if stix2.get('type', '') == 'bundle' and stix2.get('spec_version', '') == '2.0':
	objects = stix2.get('objects', [])
	u_objects = self.set_modified_time_stamp(objects, file_modified)

	if api_root not in self.cache:
	self.cache[api_root] = {'modified': '', 'files': {}}

	self.cache[api_root]['modified'] = api_root_modified
	self.cache[api_root]['files'][file_name] = {'modified': file_modified, 'objects': u_objects}
	except Exception as e:
	raise ProcessingError('error adding objects to cache', 500, e)
	finally:
	return u_objects

	def with_cache(self, api_root):
	api_root_path = os.path.join(self.path, api_root)
	api_root_modified = self.get_modified_time_stamp(api_root_path)

	if api_root in self.cache:
	if self.cache[api_root]['modified'] == api_root_modified:
	# Return objects from cache
	objects = []
	for k, v in self.cache[api_root]['files'].items():
	objects.extend(v['objects'])
	return objects
	else:
	# Cleanup the cache
	self.delete_from_cache(api_root)

	# Add to the cache and return objects for collection
	dir_list = os.listdir(api_root_path)
	files = [f for f in dir_list if os.path.isfile(os.path.join(api_root_path, f)) and f.endswith('.json')]

	objects = []
	for f in files:
	fp = os.path.join(api_root_path, f)
	file_modified = self.get_modified_time_stamp(fp)

	cached_files = self.cache[api_root]['files']
	if f in cached_files and cached_files[f]['modified'] == file_modified:
	objects.extend(cached_files[f]['objects'])
	else:
	u_objects = self.add_to_cache(api_root, api_root_modified, f, file_modified)
	objects.extend(u_objects)
	return objects
	else:
	# Update the cache and return the objects for the collection
	dir_list = os.listdir(api_root_path)
	files = [f for f in dir_list if os.path.isfile(os.path.join(api_root_path, f)) and f.endswith('.json')]

	objects = []
	for f in files:
	fp = os.path.join(api_root_path, f)
	file_modified = self.get_modified_time_stamp(fp)

	u_objects = self.add_to_cache(api_root, api_root_modified, f, file_modified)
	objects.extend(u_objects)
	return objects

	def get_objects_without_bundle(self, api_root, collection_id, filter_args, allowed_filters):
	self.update_discovery_config()

	if self.validate_requested_api_root(api_root):
	# Get the collection
	collection = None
	num_collections, collections = self.get_collections(api_root, 0, -1)

	for c in collections:
	if 'id' in c and collection_id == c['id']:
	collection = c
	break

	if not collection:
	raise ProcessingError("collection for api-root '{}' was not found".format(api_root), 500)

	# Add the objects to the collection
	collection['objects'] = self.with_cache(api_root)

	# Filter the collection
	filtered_objects = []

	if filter_args:
	full_filter = BasicFilter(filter_args)
	filtered_objects.extend(
	full_filter.process_filter(
	collection.get('objects', []),
	allowed_filters,
	collection.get('manifest', [])
	)
	)
	else:
	filtered_objects.extend(collection.get('objects', []))

	return filtered_objects

	def get_objects(self, api_root, collection_id, filter_args, allowed_filters, start_index, end_index):
	# print('start_index: {}, end_index: {}'.format(start_index, end_index))

	objects = self.get_objects_without_bundle(api_root, collection_id, filter_args, allowed_filters)

	objects.sort(key=lambda x: datetime.datetime.strptime(x['modified'], '%Y-%m-%dT%H:%M:%S.%fZ'))

	count = len(objects)

	objects = objects if end_index == -1 else objects[start_index:end_index]

	return count, create_bundle(objects)

	def get_object(self, api_root, collection_id, object_id, filter_args, allowed_filters):
	objects = self.get_objects_without_bundle(api_root, collection_id, filter_args, allowed_filters)

	req_object = [i for i in objects if i['id'] == object_id]

	if len(req_object) == 1:
	return create_bundle(req_object)

	def get_object_manifest(self, api_root, collection_id, filter_args, allowed_filters, start_index, end_index):
	self.update_discovery_config()

	if self.validate_requested_api_root(api_root):
	count, collections = self.get_collections(api_root, 0, -1)

	for collection in collections:
	if 'id' in collection and collection_id == collection['id']:
	manifest = collection.get('manifest', [])
	if filter_args:
	full_filter = BasicFilter(filter_args)
	manifest = full_filter.process_filter(
	manifest,
	allowed_filters,
	None
	)

	count = len(manifest)

	manifest = manifest if end_index == -1 else manifest[start_index:end_index]

	return count, manifest

	def add_objects(self, api_root, collection_id, objs, request_time):
	failed = 0
	succeeded = 0
	pending = 0
	successes = []
	failures = []

	file_name = '{}--{}.{}'.format(request_time, objs['id'], 'json')
	p = os.path.join(self.path, api_root)
	fp = os.path.join(p, file_name)

	try:
	add_objs = objs['objects']
	num_objs = len(add_objs)

	try:
	# Each add_object request writes the provided bundle to a new file
	with open(fp, 'w') as out_file:
	out_file.write(json.dumps(objs, indent=4, sort_keys=True))

	succeeded += num_objs
	successes = list(map(lambda x: x['id'], add_objs))

	# Update the cache after the file is written
	self.with_cache(api_root)
	except IOError:
	failed += num_objs
	failures = list(map(lambda x: x['id'], add_objs))

	except Exception as e:
	raise ProcessingError('error adding objects', 500, e)

	status = generate_status(request_time, 'complete', succeeded, failed,
	pending, successes_ids=successes, failures=failures)

	self.statuses.append(status)

	return status

	def get_status(self, api_root, status_id):
	for s in self.statuses:
	if status_id == s['id']:
	return s

oasis-open / cti-taxii-server Goto Github PK

cti-taxii-server's Introduction

cti-taxii-server

Installation

Usage

As a script

Docker

Governance

Maintainers

Current Maintainers of this TC Open Repository

About OASIS TC Open Repositories

Feedback

cti-taxii-server's People

Contributors

Stargazers

Watchers

Forkers

cti-taxii-server's Issues

Recommend Projects

Recommend Topics

Recommend Org