We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Riskiq Connector throwing errors
## Description

RiskIQ connector is not working as expected with the correct credentials defined.

## Environment

1. OS - Ubuntu 
2. OpenCTI version: 5.1.3

## Riskiq Connector Logs:
INFO:root:Listing Threat-Actors with filters null.,
INFO:root:Connector registered with ID: c455a3a4-cc8f-4133-9f8d-4098fa984de8,
INFO:root:Starting ping alive thread,
INFO:riskiq.client:URL: https://api.riskiq.net/pt/v2,
INFO:root:Starting RiskIQ connector...,
INFO:root:Running RiskIQ connector...,
INFO:root:Connector interval sec: 60,
INFO:root:[RiskIQ] loaded state: {},
INFO:root:RiskIQ connector clean run,
INFO:root:Initiate work for c455a3a4-cc8f-4133-9f8d-4098fa984de8,
INFO:root:[RiskIQ] workid opencti-work--2c314a8c-484e-4a68-9b31-bb782b3b22ed initiated,
INFO:root:[RiskIQ] last run: None,
**ERROR:root:Parser must be a string or character stream, not NoneType **

![image](https://user-images.githubusercontent.com/92973503/151704741-cd22d3ec-4963-4f7f-becd-d4930a9d0798.png)

++Config File++

  connector-riskiq:
    image: opencti/connector-riskiq:5.1.3
    environment:
      - OPENCTI_URL=http://opencti:8080
      - OPENCTI_TOKEN=c9dc7053-6bdf-44ca-9dfd-c0e3ff249eb8
      - CONNECTOR_ID=c455a3a4-cc8f-4133-9f8d-4098fa984de8
      - CONNECTOR_TYPE=EXTERNAL_IMPORT
      - CONNECTOR_NAME=RISKIQ
      - CONNECTOR_SCOPE=riskiq
      - CONNECTOR_CONFIDENCE_LEVEL=15 # From 0 (Unknown) to 100 (Fully trusted)
      - CONNECTOR_LOG_LEVEL=info
      - RISKIQ_BASE_URL=https://api.riskiq.net/pt/v2
      - RISKIQ_USER=xxx@testing.com
      - RISKIQ_PASSWORD=xxxxxxx
      - RISKIQ_INTERVAL_SEC=86400
    restart: always


It was working before, after a reboot the riskiq connector started logging the above error as "ERROR:root:Parser must be a string or character stream, not NoneType".

Please help to fix the same.

Thanks


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `external-import/riskiq/src/riskiq/article_importer.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 """OpenCTI RiskIQ's article importer module."""
3 import datetime
4 import itertools
5 from typing import Any, Mapping, Optional
6 
7 from dateutil import parser
8 from pycti import OpenCTIConnectorHelper
9 from stix2 import (
10     Bundle,
11     DomainName,
12     EmailAddress,
13     File,
14     Identity,
15     Indicator,
16     IPv4Address,
17     Mutex,
18     Report,
19     TLP_AMBER,
20     TLP_WHITE,
21     URL,
22     X509Certificate,
23     utils,
24 )
25 from stix2.v21 import _Observable
26 
27 from .utils import datetime_to_timestamp
28 
29 
30 class ArticleImporter:
31     """Article importer class."""
32 
33     _LATEST_ARTICLE_TIMESTAMP = "latest_article_timestamp"
34 
35     def __init__(
36         self, helper: OpenCTIConnectorHelper, article: dict[str, Any], author: Identity
37     ):
38         """Initialization of the article importer."""
39         self.helper = helper
40         self.article = article
41         self.author = author
42         self.work_id: Optional[str] = None
43         # Use custom properties to set the author and the confidence level of the object.
44         self.custom_props = {
45             "x_opencti_created_by_ref": self.author["id"],
46         }
47 
48     def _process_indicator(self, indicator: Indicator) -> list[_Observable]:
49         """
50         Process the indicator depending on its type.
51 
52         Parameters
53         ----------
54         indicator : Indicator
55             One indicator from an article.
56 
57         Returns
58         -------
59         List of Observable
60             A list of Observable depending on the indicator type.
61         """
62         indicator_type = indicator["type"]
63         values = indicator["values"]
64         tlp_marking = TLP_WHITE if indicator["source"] == "public" else TLP_AMBER
65 
66         if indicator_type == "hash_md5":
67             return [
68                 File(
69                     type="file",
70                     hashes={"MD5": v},
71                     object_marking_refs=tlp_marking,
72                     custom_properties=self.custom_props,
73                 )
74                 for v in values
75             ]
76 
77         if indicator_type in ["hash_sha1", "sha1"]:
78             return [
79                 File(
80                     type="file",
81                     hashes={"SHA-1": v},
82                     object_marking_refs=tlp_marking,
83                     custom_properties=self.custom_props,
84                 )
85                 for v in values
86             ]
87 
88         if indicator_type in ["sha256", "hash_sha256"]:
89             return [
90                 File(
91                     type="file",
92                     hashes={"SHA-256": v},
93                     object_marking_refs=tlp_marking,
94                     custom_properties=self.custom_props,
95                 )
96                 for v in values
97             ]
98 
99         if indicator_type == "domain":
100             return [
101                 DomainName(
102                     type="domain-name",
103                     value=v,
104                     object_marking_refs=tlp_marking,
105                     custom_properties=self.custom_props,
106                 )
107                 for v in values
108             ]
109 
110         if indicator_type in ["email", "emails"]:
111             return [
112                 EmailAddress(
113                     type="email-addr",
114                     value=v,
115                     object_marking_refs=tlp_marking,
116                     custom_properties=self.custom_props,
117                 )
118                 for v in values
119             ]
120 
121         if indicator_type in ["filename", "filepath"]:
122             return [
123                 File(
124                     type="file",
125                     name=v,
126                     object_marking_refs=tlp_marking,
127                     custom_properties=self.custom_props,
128                 )
129                 for v in values
130             ]
131 
132         if indicator_type == "ip":
133             return [
134                 IPv4Address(
135                     type="ipv4-addr",
136                     value=v,
137                     object_marking_refs=tlp_marking,
138                     custom_properties=self.custom_props,
139                 )
140                 for v in values
141             ]
142 
143         if indicator_type in ["proces_mutex", "process_mutex", "mutex"]:
144             return [
145                 Mutex(
146                     type="mutex",
147                     name=v,
148                     object_marking_refs=tlp_marking,
149                     custom_properties=self.custom_props,
150                 )
151                 for v in values
152             ]
153 
154         if indicator_type == "url":
155             return [
156                 URL(
157                     type="url",
158                     value=v,
159                     object_marking_refs=tlp_marking,
160                     defanged=False,
161                     custom_properties=self.custom_props,
162                 )
163                 for v in values
164             ]
165 
166         if indicator_type == "certificate_sha1":
167             return [
168                 X509Certificate(
169                     type="x509-certificate",
170                     hashes={"SHA-1": v},
171                     object_marking_refs=tlp_marking,
172                     custom_properties=self.custom_props,
173                 )
174                 for v in values
175             ]
176 
177         if indicator_type in [
178             "certificate_issuerorganizationname",
179             "certificate_issuercommonname",
180         ]:
181             return [
182                 X509Certificate(
183                     type="x509-certificate",
184                     issuer=v,
185                     object_marking_refs=tlp_marking,
186                     custom_properties=self.custom_props,
187                 )
188                 for v in values
189             ]
190 
191         if indicator_type in [
192             "certificate_subjectorganizationname",
193             "certificate_subjectcountry",
194             "certificate_subjectcommonname",
195         ]:
196             return [
197                 X509Certificate(
198                     type="x509-certificate",
199                     subject=v,
200                     object_marking_refs=tlp_marking,
201                     custom_properties=self.custom_props,
202                 )
203                 for v in values
204             ]
205 
206         if indicator_type in ["certificate_serialnumber", "code_certificate_serial"]:
207             return [
208                 X509Certificate(
209                     type="x509-certificate",
210                     serial_number=v,
211                     object_marking_refs=tlp_marking,
212                     custom_properties=self.custom_props,
213                 )
214                 for v in values
215             ]
216 
217         self.helper.log_warning(
218             f"[RiskIQ] indicator with key {indicator_type} not supported. (Values: {values})"
219         )
220         return []
221 
222     def run(self, work_id: str, state: Mapping[str, Any]) -> Mapping[str, Any]:
223         """Run the importation of the article."""
224         self.work_id = work_id
225         published = parser.parse(self.article["publishedDate"])
226         created = parser.parse(self.article["createdDate"])
227 
228         indicators = itertools.chain(
229             *[
230                 self._process_indicator(indicator)
231                 for indicator in self.article["indicators"]
232             ]
233         )
234 
235         indicators = utils.deduplicate(list(indicators))
236         # Return the initial state if we don't have any indicators.
237         if not indicators:
238             self.helper.log_info("No indicator in article, report will not be created.")
239             return state
240 
241         self.helper.log_debug(f"Number of indicators: {len(indicators)}")
242 
243         # Check if all indicators' TLP marking are `TLP_WHITE`.
244         report_tlp = TLP_WHITE
245         if TLP_AMBER in [i["object_marking_refs"][0] for i in indicators]:
246             report_tlp = TLP_AMBER
247 
248         report = Report(
249             type="report",
250             name=self.article.get("title", "RiskIQ Threat Report"),
251             description=self.article["summary"],
252             report_types=["threat-report"],
253             created_by_ref=self.author,
254             created=created,
255             published=published,
256             lang="en",
257             labels=self.article["tags"],
258             object_refs=indicators,
259             object_marking_refs=report_tlp,
260             external_references=[
261                 {
262                     "source_name": "riskiq",
263                     "url": self.article["link"],
264                     "external_id": self.article["guid"],
265                 }
266             ],
267             allow_custom=True,
268         )
269         self.helper.log_debug(f"[RiskIQ] Report = {report}")
270 
271         bundle = Bundle(objects=indicators + [report, self.author], allow_custom=True)
272         self.helper.log_info("[RiskIQ] Sending report STIX2 bundle")
273         self._send_bundle(bundle)
274 
275         return self._create_state(created)
276 
277     @classmethod
278     def _create_state(
279         cls, latest_datetime: Optional[datetime.datetime]
280     ) -> Mapping[str, Any]:
281         if latest_datetime is None:
282             return {}
283 
284         return {cls._LATEST_ARTICLE_TIMESTAMP: datetime_to_timestamp(latest_datetime)}
285 
286     def _send_bundle(self, bundle: Bundle) -> None:
287         serialized_bundle = bundle.serialize()
288         self.helper.send_stix2_bundle(serialized_bundle, work_id=self.work_id)
289
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/external-import/riskiq/src/riskiq/article_importer.py b/external-import/riskiq/src/riskiq/article_importer.py
--- a/external-import/riskiq/src/riskiq/article_importer.py
+++ b/external-import/riskiq/src/riskiq/article_importer.py
@@ -222,8 +222,14 @@
     def run(self, work_id: str, state: Mapping[str, Any]) -> Mapping[str, Any]:
         """Run the importation of the article."""
         self.work_id = work_id
-        published = parser.parse(self.article["publishedDate"])
         created = parser.parse(self.article["createdDate"])
+        # RisIQ API does not always provide the `publishedDate`.
+        # If it does not exist, take the value of the `createdDate` instead.
+        published = (
+            parser.parse(self.article["publishedDate"])
+            if self.article["publishedDate"] is not None
+            else created
+        )
 
         indicators = itertools.chain(
             *[

{"golden_diff": "diff --git a/external-import/riskiq/src/riskiq/article_importer.py b/external-import/riskiq/src/riskiq/article_importer.py\n--- a/external-import/riskiq/src/riskiq/article_importer.py\n+++ b/external-import/riskiq/src/riskiq/article_importer.py\n@@ -222,8 +222,14 @@\n     def run(self, work_id: str, state: Mapping[str, Any]) -> Mapping[str, Any]:\n         \"\"\"Run the importation of the article.\"\"\"\n         self.work_id = work_id\n-        published = parser.parse(self.article[\"publishedDate\"])\n         created = parser.parse(self.article[\"createdDate\"])\n+        # RisIQ API does not always provide the `publishedDate`.\n+        # If it does not exist, take the value of the `createdDate` instead.\n+        published = (\n+            parser.parse(self.article[\"publishedDate\"])\n+            if self.article[\"publishedDate\"] is not None\n+            else created\n+        )\n \n         indicators = itertools.chain(\n             *[\n", "issue": "Riskiq Connector throwing errors\n## Description\r\n\r\nRiskIQ connector is not working as expected with the correct credentials defined.\r\n\r\n## Environment\r\n\r\n1. OS - Ubuntu \r\n2. OpenCTI version: 5.1.3\r\n\r\n## Riskiq Connector Logs:\r\nINFO:root:Listing Threat-Actors with filters null.,\r\nINFO:root:Connector registered with ID: c455a3a4-cc8f-4133-9f8d-4098fa984de8,\r\nINFO:root:Starting ping alive thread,\r\nINFO:riskiq.client:URL: https://api.riskiq.net/pt/v2,\r\nINFO:root:Starting RiskIQ connector...,\r\nINFO:root:Running RiskIQ connector...,\r\nINFO:root:Connector interval sec: 60,\r\nINFO:root:[RiskIQ] loaded state: {},\r\nINFO:root:RiskIQ connector clean run,\r\nINFO:root:Initiate work for c455a3a4-cc8f-4133-9f8d-4098fa984de8,\r\nINFO:root:[RiskIQ] workid opencti-work--2c314a8c-484e-4a68-9b31-bb782b3b22ed initiated,\r\nINFO:root:[RiskIQ] last run: None,\r\n**ERROR:root:Parser must be a string or character stream, not NoneType **\r\n\r\n![image](https://user-images.githubusercontent.com/92973503/151704741-cd22d3ec-4963-4f7f-becd-d4930a9d0798.png)\r\n\r\n++Config File++\r\n\r\n  connector-riskiq:\r\n    image: opencti/connector-riskiq:5.1.3\r\n    environment:\r\n      - OPENCTI_URL=http://opencti:8080\r\n      - OPENCTI_TOKEN=c9dc7053-6bdf-44ca-9dfd-c0e3ff249eb8\r\n      - CONNECTOR_ID=c455a3a4-cc8f-4133-9f8d-4098fa984de8\r\n      - CONNECTOR_TYPE=EXTERNAL_IMPORT\r\n      - CONNECTOR_NAME=RISKIQ\r\n      - CONNECTOR_SCOPE=riskiq\r\n      - CONNECTOR_CONFIDENCE_LEVEL=15 # From 0 (Unknown) to 100 (Fully trusted)\r\n      - CONNECTOR_LOG_LEVEL=info\r\n      - RISKIQ_BASE_URL=https://api.riskiq.net/pt/v2\r\n      - RISKIQ_USER=xxx@testing.com\r\n      - RISKIQ_PASSWORD=xxxxxxx\r\n      - RISKIQ_INTERVAL_SEC=86400\r\n    restart: always\r\n\r\n\r\nIt was working before, after a reboot the riskiq connector started logging the above error as \"ERROR:root:Parser must be a string or character stream, not NoneType\".\r\n\r\nPlease help to fix the same.\r\n\r\nThanks\r\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"OpenCTI RiskIQ's article importer module.\"\"\"\nimport datetime\nimport itertools\nfrom typing import Any, Mapping, Optional\n\nfrom dateutil import parser\nfrom pycti import OpenCTIConnectorHelper\nfrom stix2 import (\n    Bundle,\n    DomainName,\n    EmailAddress,\n    File,\n    Identity,\n    Indicator,\n    IPv4Address,\n    Mutex,\n    Report,\n    TLP_AMBER,\n    TLP_WHITE,\n    URL,\n    X509Certificate,\n    utils,\n)\nfrom stix2.v21 import _Observable\n\nfrom .utils import datetime_to_timestamp\n\n\nclass ArticleImporter:\n    \"\"\"Article importer class.\"\"\"\n\n    _LATEST_ARTICLE_TIMESTAMP = \"latest_article_timestamp\"\n\n    def __init__(\n        self, helper: OpenCTIConnectorHelper, article: dict[str, Any], author: Identity\n    ):\n        \"\"\"Initialization of the article importer.\"\"\"\n        self.helper = helper\n        self.article = article\n        self.author = author\n        self.work_id: Optional[str] = None\n        # Use custom properties to set the author and the confidence level of the object.\n        self.custom_props = {\n            \"x_opencti_created_by_ref\": self.author[\"id\"],\n        }\n\n    def _process_indicator(self, indicator: Indicator) -> list[_Observable]:\n        \"\"\"\n        Process the indicator depending on its type.\n\n        Parameters\n        ----------\n        indicator : Indicator\n            One indicator from an article.\n\n        Returns\n        -------\n        List of Observable\n            A list of Observable depending on the indicator type.\n        \"\"\"\n        indicator_type = indicator[\"type\"]\n        values = indicator[\"values\"]\n        tlp_marking = TLP_WHITE if indicator[\"source\"] == \"public\" else TLP_AMBER\n\n        if indicator_type == \"hash_md5\":\n            return [\n                File(\n                    type=\"file\",\n                    hashes={\"MD5\": v},\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"hash_sha1\", \"sha1\"]:\n            return [\n                File(\n                    type=\"file\",\n                    hashes={\"SHA-1\": v},\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"sha256\", \"hash_sha256\"]:\n            return [\n                File(\n                    type=\"file\",\n                    hashes={\"SHA-256\": v},\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type == \"domain\":\n            return [\n                DomainName(\n                    type=\"domain-name\",\n                    value=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"email\", \"emails\"]:\n            return [\n                EmailAddress(\n                    type=\"email-addr\",\n                    value=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"filename\", \"filepath\"]:\n            return [\n                File(\n                    type=\"file\",\n                    name=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type == \"ip\":\n            return [\n                IPv4Address(\n                    type=\"ipv4-addr\",\n                    value=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"proces_mutex\", \"process_mutex\", \"mutex\"]:\n            return [\n                Mutex(\n                    type=\"mutex\",\n                    name=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type == \"url\":\n            return [\n                URL(\n                    type=\"url\",\n                    value=v,\n                    object_marking_refs=tlp_marking,\n                    defanged=False,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type == \"certificate_sha1\":\n            return [\n                X509Certificate(\n                    type=\"x509-certificate\",\n                    hashes={\"SHA-1\": v},\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\n            \"certificate_issuerorganizationname\",\n            \"certificate_issuercommonname\",\n        ]:\n            return [\n                X509Certificate(\n                    type=\"x509-certificate\",\n                    issuer=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\n            \"certificate_subjectorganizationname\",\n            \"certificate_subjectcountry\",\n            \"certificate_subjectcommonname\",\n        ]:\n            return [\n                X509Certificate(\n                    type=\"x509-certificate\",\n                    subject=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"certificate_serialnumber\", \"code_certificate_serial\"]:\n            return [\n                X509Certificate(\n                    type=\"x509-certificate\",\n                    serial_number=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        self.helper.log_warning(\n            f\"[RiskIQ] indicator with key {indicator_type} not supported. (Values: {values})\"\n        )\n        return []\n\n    def run(self, work_id: str, state: Mapping[str, Any]) -> Mapping[str, Any]:\n        \"\"\"Run the importation of the article.\"\"\"\n        self.work_id = work_id\n        published = parser.parse(self.article[\"publishedDate\"])\n        created = parser.parse(self.article[\"createdDate\"])\n\n        indicators = itertools.chain(\n            *[\n                self._process_indicator(indicator)\n                for indicator in self.article[\"indicators\"]\n            ]\n        )\n\n        indicators = utils.deduplicate(list(indicators))\n        # Return the initial state if we don't have any indicators.\n        if not indicators:\n            self.helper.log_info(\"No indicator in article, report will not be created.\")\n            return state\n\n        self.helper.log_debug(f\"Number of indicators: {len(indicators)}\")\n\n        # Check if all indicators' TLP marking are `TLP_WHITE`.\n        report_tlp = TLP_WHITE\n        if TLP_AMBER in [i[\"object_marking_refs\"][0] for i in indicators]:\n            report_tlp = TLP_AMBER\n\n        report = Report(\n            type=\"report\",\n            name=self.article.get(\"title\", \"RiskIQ Threat Report\"),\n            description=self.article[\"summary\"],\n            report_types=[\"threat-report\"],\n            created_by_ref=self.author,\n            created=created,\n            published=published,\n            lang=\"en\",\n            labels=self.article[\"tags\"],\n            object_refs=indicators,\n            object_marking_refs=report_tlp,\n            external_references=[\n                {\n                    \"source_name\": \"riskiq\",\n                    \"url\": self.article[\"link\"],\n                    \"external_id\": self.article[\"guid\"],\n                }\n            ],\n            allow_custom=True,\n        )\n        self.helper.log_debug(f\"[RiskIQ] Report = {report}\")\n\n        bundle = Bundle(objects=indicators + [report, self.author], allow_custom=True)\n        self.helper.log_info(\"[RiskIQ] Sending report STIX2 bundle\")\n        self._send_bundle(bundle)\n\n        return self._create_state(created)\n\n    @classmethod\n    def _create_state(\n        cls, latest_datetime: Optional[datetime.datetime]\n    ) -> Mapping[str, Any]:\n        if latest_datetime is None:\n            return {}\n\n        return {cls._LATEST_ARTICLE_TIMESTAMP: datetime_to_timestamp(latest_datetime)}\n\n    def _send_bundle(self, bundle: Bundle) -> None:\n        serialized_bundle = bundle.serialize()\n        self.helper.send_stix2_bundle(serialized_bundle, work_id=self.work_id)\n", "path": "external-import/riskiq/src/riskiq/article_importer.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"OpenCTI RiskIQ's article importer module.\"\"\"\nimport datetime\nimport itertools\nfrom typing import Any, Mapping, Optional\n\nfrom dateutil import parser\nfrom pycti import OpenCTIConnectorHelper\nfrom stix2 import (\n    Bundle,\n    DomainName,\n    EmailAddress,\n    File,\n    Identity,\n    Indicator,\n    IPv4Address,\n    Mutex,\n    Report,\n    TLP_AMBER,\n    TLP_WHITE,\n    URL,\n    X509Certificate,\n    utils,\n)\nfrom stix2.v21 import _Observable\n\nfrom .utils import datetime_to_timestamp\n\n\nclass ArticleImporter:\n    \"\"\"Article importer class.\"\"\"\n\n    _LATEST_ARTICLE_TIMESTAMP = \"latest_article_timestamp\"\n\n    def __init__(\n        self, helper: OpenCTIConnectorHelper, article: dict[str, Any], author: Identity\n    ):\n        \"\"\"Initialization of the article importer.\"\"\"\n        self.helper = helper\n        self.article = article\n        self.author = author\n        self.work_id: Optional[str] = None\n        # Use custom properties to set the author and the confidence level of the object.\n        self.custom_props = {\n            \"x_opencti_created_by_ref\": self.author[\"id\"],\n        }\n\n    def _process_indicator(self, indicator: Indicator) -> list[_Observable]:\n        \"\"\"\n        Process the indicator depending on its type.\n\n        Parameters\n        ----------\n        indicator : Indicator\n            One indicator from an article.\n\n        Returns\n        -------\n        List of Observable\n            A list of Observable depending on the indicator type.\n        \"\"\"\n        indicator_type = indicator[\"type\"]\n        values = indicator[\"values\"]\n        tlp_marking = TLP_WHITE if indicator[\"source\"] == \"public\" else TLP_AMBER\n\n        if indicator_type == \"hash_md5\":\n            return [\n                File(\n                    type=\"file\",\n                    hashes={\"MD5\": v},\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"hash_sha1\", \"sha1\"]:\n            return [\n                File(\n                    type=\"file\",\n                    hashes={\"SHA-1\": v},\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"sha256\", \"hash_sha256\"]:\n            return [\n                File(\n                    type=\"file\",\n                    hashes={\"SHA-256\": v},\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type == \"domain\":\n            return [\n                DomainName(\n                    type=\"domain-name\",\n                    value=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"email\", \"emails\"]:\n            return [\n                EmailAddress(\n                    type=\"email-addr\",\n                    value=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"filename\", \"filepath\"]:\n            return [\n                File(\n                    type=\"file\",\n                    name=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type == \"ip\":\n            return [\n                IPv4Address(\n                    type=\"ipv4-addr\",\n                    value=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"proces_mutex\", \"process_mutex\", \"mutex\"]:\n            return [\n                Mutex(\n                    type=\"mutex\",\n                    name=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type == \"url\":\n            return [\n                URL(\n                    type=\"url\",\n                    value=v,\n                    object_marking_refs=tlp_marking,\n                    defanged=False,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type == \"certificate_sha1\":\n            return [\n                X509Certificate(\n                    type=\"x509-certificate\",\n                    hashes={\"SHA-1\": v},\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\n            \"certificate_issuerorganizationname\",\n            \"certificate_issuercommonname\",\n        ]:\n            return [\n                X509Certificate(\n                    type=\"x509-certificate\",\n                    issuer=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\n            \"certificate_subjectorganizationname\",\n            \"certificate_subjectcountry\",\n            \"certificate_subjectcommonname\",\n        ]:\n            return [\n                X509Certificate(\n                    type=\"x509-certificate\",\n                    subject=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        if indicator_type in [\"certificate_serialnumber\", \"code_certificate_serial\"]:\n            return [\n                X509Certificate(\n                    type=\"x509-certificate\",\n                    serial_number=v,\n                    object_marking_refs=tlp_marking,\n                    custom_properties=self.custom_props,\n                )\n                for v in values\n            ]\n\n        self.helper.log_warning(\n            f\"[RiskIQ] indicator with key {indicator_type} not supported. (Values: {values})\"\n        )\n        return []\n\n    def run(self, work_id: str, state: Mapping[str, Any]) -> Mapping[str, Any]:\n        \"\"\"Run the importation of the article.\"\"\"\n        self.work_id = work_id\n        created = parser.parse(self.article[\"createdDate\"])\n        # RisIQ API does not always provide the `publishedDate`.\n        # If it does not exist, take the value of the `createdDate` instead.\n        published = (\n            parser.parse(self.article[\"publishedDate\"])\n            if self.article[\"publishedDate\"] is not None\n            else created\n        )\n\n        indicators = itertools.chain(\n            *[\n                self._process_indicator(indicator)\n                for indicator in self.article[\"indicators\"]\n            ]\n        )\n\n        indicators = utils.deduplicate(list(indicators))\n        # Return the initial state if we don't have any indicators.\n        if not indicators:\n            self.helper.log_info(\"No indicator in article, report will not be created.\")\n            return state\n\n        self.helper.log_debug(f\"Number of indicators: {len(indicators)}\")\n\n        # Check if all indicators' TLP marking are `TLP_WHITE`.\n        report_tlp = TLP_WHITE\n        if TLP_AMBER in [i[\"object_marking_refs\"][0] for i in indicators]:\n            report_tlp = TLP_AMBER\n\n        report = Report(\n            type=\"report\",\n            name=self.article.get(\"title\", \"RiskIQ Threat Report\"),\n            description=self.article[\"summary\"],\n            report_types=[\"threat-report\"],\n            created_by_ref=self.author,\n            created=created,\n            published=published,\n            lang=\"en\",\n            labels=self.article[\"tags\"],\n            object_refs=indicators,\n            object_marking_refs=report_tlp,\n            external_references=[\n                {\n                    \"source_name\": \"riskiq\",\n                    \"url\": self.article[\"link\"],\n                    \"external_id\": self.article[\"guid\"],\n                }\n            ],\n            allow_custom=True,\n        )\n        self.helper.log_debug(f\"[RiskIQ] Report = {report}\")\n\n        bundle = Bundle(objects=indicators + [report, self.author], allow_custom=True)\n        self.helper.log_info(\"[RiskIQ] Sending report STIX2 bundle\")\n        self._send_bundle(bundle)\n\n        return self._create_state(created)\n\n    @classmethod\n    def _create_state(\n        cls, latest_datetime: Optional[datetime.datetime]\n    ) -> Mapping[str, Any]:\n        if latest_datetime is None:\n            return {}\n\n        return {cls._LATEST_ARTICLE_TIMESTAMP: datetime_to_timestamp(latest_datetime)}\n\n    def _send_bundle(self, bundle: Bundle) -> None:\n        serialized_bundle = bundle.serialize()\n        self.helper.send_stix2_bundle(serialized_bundle, work_id=self.work_id)\n", "path": "external-import/riskiq/src/riskiq/article_importer.py"}]}