Merge branch 'release/0.4.2'

README.md

@@ -1,7 +1,10 @@
 # Common OSINT Model
 
 **Note:** This is work in progress and probably only covers my specific use case. If you find bugs or know how to
-enhance this project, please open an issue or - even better - create a pull request.
+enhance this project, please open an issue or - even better - create a pull request. The presented data model is
+evolving continuously. Therefore, it is recommended to use it in your project with a fixed version constraint (e.g. 
+`common-osint-model==0.4.1`) and take a look at what has changed here before updating `common-osint-model` as a
+dependency.
 
 This project aims to create an easy to use data model as well as implement converters for commonly used sources. As my
 use case often includes HTTP(S), TLS and SSH only, data delivered for other protocols by the given sources might not

common_osint_model/__init__.py

@@ -9,7 +9,7 @@
 from common_osint_model.certificate import from_x509_pem, from_x509_pem_flattened
 
 from common_osint_model.models.host import Host
-from common_osint_model.models.domain import Domain
+from common_osint_model.models.domain import *
 from common_osint_model.models.service import Service
 from common_osint_model.models.autonomous_system import AutonomousSystem
 from common_osint_model.models.http import *

common_osint_model/models/domain.py

@@ -1,13 +1,30 @@
 from datetime import datetime
-from typing import Optional
+from typing import Optional, List
 
 from pydantic import BaseModel
 
 
+class Entity(BaseModel):
+    """Represents an entity which registered a domain."""
+    name: Optional[str]
+    email: Optional[str]
+    organization: Optional[str]
+    street: Optional[str]
+    city: Optional[str]
+    postal_code: Optional[str]
+    country: Optional[str]
+    phone: Optional[str]
+    timestamp: datetime = datetime.utcnow()
+
+
 class Domain(BaseModel):
     """Represents a domain pointing to a specific host."""
     domain: str
     first_seen: datetime = datetime.utcnow()
     last_seen: datetime = datetime.utcnow()
     source: Optional[str]
     type: Optional[str]
+    soa: Optional[List[str]]
+    nameserver: Optional[List[str]]
+    registrar: Optional[str]
+    registrant: Optional[Entity]

common_osint_model/models/host.py

@@ -15,11 +15,22 @@
 class Host(BaseModel, ShodanDataHandler, CensysDataHandler, BinaryEdgeDataHandler, Logger):
     """This class represents a host and can be used to handle results from the common model in a pythonic way."""
     ip: str
+    # Information about the autonomous system the host is assigned to
     autonomous_system: Optional[AutonomousSystem]
-    services: List[Service]
+    # List of services running (listening) on the IP
+    services: Optional[List[Service]]
+    # List of open ports also mentioned in the open
+    ports: Optional[List[int]]
+    # Timestamps for activity tracking
     first_seen: Optional[datetime] = datetime.utcnow()
     last_seen: Optional[datetime] = datetime.utcnow()
+    # A list of domains, fqdns, common names - or other attributes which represent domainnames -  assigned to the host
     domains: Optional[List[Domain]]
+    # This represents the source where the host information was obtained, e.g. shodan, censys...
+    source: Optional[str]
+    # Optionally, the used query to find the host can be assigned to the object also which might be useful for comparing
+    # different hosts later on
+    query: Optional[str]
 
     @validator("ip")
     def validates_ip(cls, v):
@@ -35,15 +46,20 @@ def services_dict(self):
         """Returns the services as dictionary in the form of {port: service}. Uses exclude_none to skip empty keys."""
         # Load the JSON dump, so datetime objects are in iso format.
         json_dict = json.loads(self.json(exclude_none=True))
-        return {s["port"]: s for s in json_dict["services"]}
+        json_dict.update({s["port"]: s for s in json_dict["services"]})
+        del json_dict["services"]
+        return json_dict
 
     @property
     def flattened_dict(self):
         """Dict in the flattened format."""
         return flatten(self.services_dict)
 
     @property
-    def ports(self):
+    def service_ports(self):
+        """Dynamic attribute which loops over available services and grabs the port number. This is kind of redundant
+        to the ports attribute, if given, but can help to easily get the values needed for the attribute. Unfortunately
+        Pydantic does not support these kind of properties in the data model right now."""
         return [service.port for service in self.services]
 
     def flattened_json(self) -> str:
@@ -91,7 +107,9 @@ def from_shodan(cls, d: Dict):
             ip=ip,
             autonomous_system=autonomous_system,
             services=services,
-            domains=domains
+            domains=domains,
+            source="shodan",
+            ports=[service.port for service in services]
         )
 
     @classmethod
@@ -120,7 +138,9 @@ def from_censys(cls, d: Dict):
             ip=ip,
             autonomous_system=AutonomousSystem.from_censys(d),
             services=services,
-            domains=domains
+            domains=domains,
+            source="censys",
+            ports=[service.port for service in services]
         )
 
     @classmethod
@@ -129,6 +149,8 @@ def from_binaryedge(cls, d: Union[Dict, List]):
         if isinstance(d, Dict) and "results" in d:
             # This is a complete result dictionary, extract the list of services.
             d = d["results"][list(d["results"].keys())[0]]
+        elif isinstance(d, Dict) and "events" in d:
+            d = d["events"]
 
         services = {}
         for service in d:
@@ -156,5 +178,7 @@ def from_binaryedge(cls, d: Union[Dict, List]):
         return Host(
             ip=ip,
             services=services_objects,
-            domains=domains
+            domains=domains,
+            source="binaryedge",
+            ports=[service.port for service in services]
         )

common_osint_model/models/service.py

@@ -143,7 +143,8 @@ def from_binaryedge(cls, d: List):
         banner = None
         md5, sha1, sha256, murmur = None, None, None, None
         if "service-simple" in type_index:
-            banner = d[type_index["service-simple"]]["result"]["data"]["service"]["banner"]
+            banner = d[type_index["service-simple"]]["result"]["data"]["service"].get("banner", None)
+        if banner:
             md5, sha1, sha256, murmur = hash_all(banner.encode("utf-8"))
 
         return Service(

common_osint_model/models/tls.py

@@ -264,7 +264,7 @@ def from_censys(cls, d: Dict):
             issued=None,
             expires=None,
             expired=None,
-            alternative_names=d["names"],
+            alternative_names=d.get("names", None),
             sha256=d["fingerprint"]
         )