Schemas validation and easy creation

[udgover]

Important

This PR removes add_text method in Observable class. It breaks private plugins relying on this method to create observables.

New features

Observables

Validators

Overview

Validators are now enforced for observables listed below. Validators ensure that provided value is relevant and that an for example an IPv4 observable cannot be built from a Url.

• bic
• email
• hostname
• iban
• ipv4
• ipv6
• mac_address
• md5
• path
• sha1
• sha256
• url

Note

Validators enforcement also impacts url observable creation which requires a scheme to be valid. This means that example.com/foobar is no longer a valid url.

Implementation details

Value validation relies on pydandic field_validator decorator. Any observable implementation that requires validation must implement validate_value class method with @field_validator("value") decorator. validate_value is also where refang of the value takes place. If the value is not valid, ValueError exception must be raised. Otherwise, the value (modified or not) must be returned as a string.

class IPv4(observable.Observable):
    type: Literal["ipv4"] = "ipv4"

    @field_validator("value")
    @classmethod
    def validate_value(cls, value: str) -> str:
        value = observable.refang(value)
        if not validators.ipv4(value):
            raise ValueError("Invalid ipv4 address")
        return value

Helper functions

To ease observables creation, new functions have been implemented in observable module:

>>> from core.schemas import observable
>>> observable.guess_type("192.168.1[.]2")
'ipv4'
>>> observable.create(value="192.168.1[.]2")
IPv4(value='192.168.1.2', created=datetime.datetime(2024, 10, 29, 23, 57, 4, 504655, tzinfo=datetime.timezone.utc), context=[], last_analysis={}, type='ipv4', id=None, tags={}, root_type='observable')
>>> observable.find(value="192.168.1[.]2")
>>> observable.save(value="192.168.1[.]2")
IPv4(value='192.168.1.2', created=datetime.datetime(2024, 10, 29, 23, 57, 20, 82795, tzinfo=TzInfo(UTC)), context=[], last_analysis={}, type='ipv4', id='5638348', tags={}, root_type='observable')
>>> observable.find(value="192.168.1[.]2")
IPv4(value='192.168.1.2', created=datetime.datetime(2024, 10, 29, 23, 57, 20, 82795, tzinfo=TzInfo(UTC)), context=[], last_analysis={}, type='ipv4', id='5638348', tags={}, root_type='observable')

Guess observable type from value

guess_type(value: str)

Guess which observable type can validate value string. If the value can be validated by an observable, its type is returned as a string. If the value can't be guessed, None is returned.

Create an observable from string

Create an observable from the given string value without saving it in the database. It's up to the caller to then call save() on the returned object to save it in the database.

create(value: str, type: str | None = None, **kwargs)

• Params

  • value: a string defining the observable.
  • type: an optional string defining the type of the observable. If type is not provided, it will be guessed by calling guess_type.
  • kwargs: a dictionary defining further attributes depending on the type of the observable.

• Returns
  If the function call succeeds, the created observable is returned. Otherwise, an ValueError exception is raised.

Save an observable from string

Use create function and save observable in the database. Tag the resulting observable with the list of tags if defined.

save(*, value: str, type: str | None = None, tags: List[str] = None, **kwargs) -> ObservableTypes

• Params

  • value: a string defining the observable.
  • type: an optional string defining the type of the observable. If type is not provided, it will be guessed by calling guess_type.
  • tags: an optional list of strings corresponding to the tags to add to the saved observable.
  • kwargs: a dictionary defining further attributes depending on the type of the observable.

• Returns
  If the function call succeeds, the saved observable is returned. Otherwise, an ValueError exception is raised.

Create observables from text

Create observables from the given text. Each line must contain one observable. Guess the type of the observable and create it. Created observable are not saved in the database.

create_from_text(text: str)

• Params
  • text: a string containing one observable per line.
• Returns
  A tuple containing a list of created observables and a list of string for unguessable lines.

Save observables from text

Use create_from_text function and save observables in the database. Tag the resulting observables with the list of tags if defined.

save_from_text(text: str, tags: List[str] = None)

• Params
  • text: a string containing one observable per line.
  • tags: an optional list of strings corresponding to the tags to add to the saved observables.
• Returns
  A tuple containing a list of saved observables and a list of string for unguessable lines.

Create observables from file

Create observables from the given file. Each line must contain one observable. Guess the type of the observable and creates it. Created observables are not saved in the database.

create_from_file(file: FileLikeObject)

• Params
  • file: Represents a file in different ways:
    • str, bytes or os.PathLike: the provided file path will be opened for reading lines
    • io.IOBase: any file object either created from open or based on io classes
    • tempfile.SpooledTemporaryFile: defined to support API endpoint from UploadFile FastAPI.
• Returns
  A tuple containing a list of created observables and a list of string for unguessable lines.

Save observables from file

Use create_from_file function and save observables into the database. Tag the resulting observables with the list of tags if defined.

save_from_file(file: FileLikeObject, tags: List[str] = None)

• Params
  • file: Represents a file in different ways:
    • str, bytes or os.PathLike: the provided file path will be opened for reading lines
    • io.IOBase: any file object either created from open or based on io classes
    • tempfile.SpooledTemporaryFile: defined to support API endpoint from UploadFile FastAPI.
  • tags: an optional list of strings corresponding to the tags to add to the saved observables.
• Returns
  A tuple containing a list of saved observables and a list of string for unguessable lines.

Create observables from a url

Create observables from the given url. Each line must contain one observable. Guess the type of the observable and create it. Created observables are not saved in the database.

create_from_url(url: str)

• Params
  • url: a string defining the URL to fetch the content from.
• Returns
  A tuple containing a list of created observables and a list of string for unguessable lines.

Save observables from a url

Use create_from_url function and save observables in the database. Tag the resulting observabls with the list of tags if defined.

save_from_url(url: str, tags: List[str] = None)

• Params
  • url: a string defining the URL to fetch the content from.
  • tags: an optional list of strings corresponding to the tags to add to the saved observables.
• Returns
  A tuple containing a list of saved observables and a list of string for unguessable lines.

Find an observable object

find(value: str, **kwargs)

Find an observable in the database matching the string value and optional fields represented by kwargs. This function automatically refangs defanged value before querying the database. Return an observable object.

Entities

Create an entity

Create an entity from the given name and type without saving it to the database.

create(name: str, type: str, **kwargs)

• Params
  • name: a string defining the name of the entity
  • type: a string defining the type of the entity
  • kwargs: a dictionary defining further attributes depending on the type of the entity.
• Returns
  If the function call succeeds, the created entity is returned. Otherwise, an ValueError exception is raised.

Save an entity

save(name: str, type: str, tags: List[str] = None, **kwargs)

Use create function and save entity in the database. Tag the resulting entity with the list of tags if defined.

• Params
  • name: a string defining the name of the entity to create
  • type: a string defining the type of the entity to create
  • tags: an optional list of strings corresponding to the tags to add to the saved observables.
  • kwargs: a dictionary defining further attributes depending on the type of the entity.
• Returns
  If the function call succeeds, the saved entity is returned. Otherwise, an ValueError exception is raised.

Indicators

Create an indicator

Create an indicator from the given name, type, pattern and diamond without saving it to the database.

create(name: str, type: str, pattern: str, diamond: DiamondModel, **kwargs)

• Params
  • name: a string defining the name of the indicator
  • type: a string defining the type of the indicator
  • pattern: a string defining the pattern of the indicator
  • diamond a string defining the diamond model of the indicator
  • kwargs: a dictionary defining further attributes depending on the type of the indicator.
• Returns
  If the function call succeeds, the created indicator is returned. Otherwise, an ValueError exception is raised.

Save an entity

save(name: str, type: str, pattern: str, diamond: DiamondModel, tags: List[str] = None, **kwargs)

Use create function and save indicator in the database. Tag the resulting indicator with the list of tags if defined.

• Params
  • name: a string defining the name of the indicator
  • type: a string defining the type of the indicator
  • pattern: a string defining the pattern of the indicator
  • diamond a string defining the diamond model of the indicator
  • tags: an optional list of strings corresponding to the tags to add to the saved observables.
  • kwargs: a dictionary defining further attributes depending on the type of the entity.
• Returns
  If the function call succeeds, the saved indicator is returned. Otherwise, an ValueError exception is raised.

API

New observables endpoints have been implemented to support the new save and create functions:

Import observables from text

Save observables from a text containing one observable per line. Tag with defined tags.

• Endpoint POST api/v2/observables/import/text
• Usage:

import requests

observables = """1.1.1[.]1
8.8.8.8
tomchop[.]me
google.com
http://google.com/
http://tomchop[.]me/
d41d8cd98f00b204e9800998ecf8427e
da39a3ee5e6b4b0d3255bfef95601890afd80709
e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"""

requests.post("api/v2/observables/import/text", json={"text": observables, "tags": ["tag1", "tag2"]})

Import observables from file

Save observables from a file containing one observable per line. Tag with defined tags.

• Endpoint POST api/v2/observables/import/file
• Usage:

import requests

with open(path_to_file, "rb") as file:
   requests.post("/api/v2/observables/import/file", files={"file": file}, data={"tags": ["tag1", "tag2"]})

Import observables from URL

Save observables from a text containing one observable per line. Tag with defined tags.

• Endpoint POST api/v2/observables/import/file
• Usage:

import requests

requests.post("api/v2/observables/import/text",  json={"url": "https://example.com", "tags": ["tag1", "tag2"]})

[tomchop]

Some minor comments, but otherwise this looks great!

[udgover]

Fixed your comments

[tomchop]

Minor nits, otherwise looks good!