forked from elastic/connectors
-
Notifications
You must be signed in to change notification settings - Fork 0
/
confluence.py
1233 lines (1103 loc) · 47.6 KB
/
confluence.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#
# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
# or more contributor license agreements. Licensed under the Elastic License 2.0;
# you may not use this file except in compliance with the Elastic License 2.0.
#
"""Confluence source module responsible to fetch documents from Confluence Cloud/Server.
"""
import asyncio
import os
from copy import copy
from functools import partial
from urllib.parse import urljoin
import aiohttp
from aiohttp.client_exceptions import ClientResponseError, ServerDisconnectedError
from connectors.access_control import ACCESS_CONTROL
from connectors.logger import logger
from connectors.source import BaseDataSource, ConfigurableFieldValueError
from connectors.sources.atlassian import (
AtlassianAccessControl,
AtlassianAdvancedRulesValidator,
prefix_account_email,
prefix_account_id,
prefix_account_name,
prefix_group,
prefix_group_id,
prefix_user,
)
from connectors.utils import (
CancellableSleeps,
ConcurrentTasks,
MemQueue,
RetryStrategy,
iso_utc,
retryable,
ssl_context,
)
RETRIES = 3
RETRY_INTERVAL = 2
DEFAULT_RETRY_SECONDS = 30
LIMIT = 100
SPACE = "space"
SPACE_PERMISSION = "space_permission"
BLOGPOST = "blogpost"
PAGE = "page"
ATTACHMENT = "attachment"
CONTENT = "content"
DOWNLOAD = "download"
SEARCH = "search"
USER = "user"
USERS_FOR_DATA_CENTER = "users_for_data_center"
SEARCH_FOR_DATA_CENTER = "search_for_data_center"
USERS_FOR_SERVER = "users_for_server"
SPACE_QUERY = "limit=100&expand=permissions"
ATTACHMENT_QUERY = "limit=100&expand=version"
CONTENT_QUERY = "limit=50&expand=ancestors,children.attachment,history.lastUpdated,body.storage,space,space.permissions,restrictions.read.restrictions.user,restrictions.read.restrictions.group"
SEARCH_QUERY = "limit=100&expand=content.extensions,content.container,content.space,space.description"
USER_QUERY = "expand=groups,applicationRoles"
LABEL = "label"
URLS = {
SPACE: "rest/api/space?{api_query}",
SPACE_PERMISSION: "rest/extender/1.0/permission/space/{space_key}/getSpacePermissionActors/VIEWSPACE",
CONTENT: "rest/api/content/search?{api_query}",
ATTACHMENT: "rest/api/content/{id}/child/attachment?{api_query}",
SEARCH: "rest/api/search?cql={query}",
SEARCH_FOR_DATA_CENTER: "rest/api/search?cql={query}&start={start}",
USER: "rest/api/3/users/search",
USERS_FOR_DATA_CENTER: "rest/api/user/list?limit={limit}&start={start}",
USERS_FOR_SERVER: "rest/extender/1.0/user/getUsersWithConfluenceAccess?showExtendedDetails=true&startAt={start}&maxResults={limit}",
LABEL: "rest/api/content/{id}/label",
}
PING_URL = "rest/api/space?limit=1"
MAX_CONCURRENT_DOWNLOADS = 50 # Max concurrent download supported by confluence
MAX_CONCURRENCY = 50
QUEUE_SIZE = 1024
QUEUE_MEM_SIZE = 25 * 1024 * 1024 # Size in Megabytes
SERVER_USER_BATCH = 1000
DATACENTER_USER_BATCH = 200
END_SIGNAL = "FINISHED_TASK"
CONFLUENCE_CLOUD = "confluence_cloud"
CONFLUENCE_SERVER = "confluence_server"
CONFLUENCE_DATA_CENTER = "confluence_data_center"
WILDCARD = "*"
class ThrottledError(Exception):
"""Internal exception class to indicate that request was throttled by the API"""
pass
class InternalServerError(Exception):
pass
class NotFound(Exception):
pass
class ConfluenceClient:
"""Confluence client to handle API calls made to Confluence"""
def __init__(self, configuration):
self._sleeps = CancellableSleeps()
self.configuration = configuration
self._logger = logger
self.data_source_type = self.configuration["data_source"]
self.host_url = self.configuration["confluence_url"]
self.ssl_enabled = self.configuration["ssl_enabled"]
self.certificate = self.configuration["ssl_ca"]
self.retry_count = self.configuration["retry_count"]
self.index_labels = self.configuration["index_labels"]
if self.data_source_type == CONFLUENCE_CLOUD:
self.host_url = os.path.join(self.host_url, "wiki")
if self.ssl_enabled and self.certificate:
self.ssl_ctx = ssl_context(certificate=self.certificate)
else:
self.ssl_ctx = False
self.session = None
def set_logger(self, logger_):
self._logger = logger_
def _get_session(self):
"""Generate and return base client session with configuration fields
Returns:
aiohttp.ClientSession: An instance of Client Session
"""
if self.session:
return self.session
self._logger.debug("Creating a client session")
if self.data_source_type == CONFLUENCE_CLOUD:
auth = (
self.configuration["account_email"],
self.configuration["api_token"],
)
elif self.data_source_type == CONFLUENCE_SERVER:
auth = (
self.configuration["username"],
self.configuration["password"],
)
else:
auth = (
self.configuration["data_center_username"],
self.configuration["data_center_password"],
)
basic_auth = aiohttp.BasicAuth(login=auth[0], password=auth[1])
timeout = aiohttp.ClientTimeout(total=None) # pyright: ignore
self.session = aiohttp.ClientSession(
auth=basic_auth,
headers={
"accept": "application/json",
"content-type": "application/json",
},
timeout=timeout,
raise_for_status=True,
)
return self.session
async def close_session(self):
"""Closes unclosed client session"""
self._sleeps.cancel()
if self.session is None:
return
await self.session.close()
self.session = None
async def _handle_client_errors(self, url, exception):
if exception.status == 429:
response_headers = exception.headers or {}
retry_seconds = DEFAULT_RETRY_SECONDS
if "Retry-After" in response_headers:
try:
retry_seconds = int(response_headers["Retry-After"])
except (TypeError, ValueError) as exception:
self._logger.error(
f"Error while reading value of retry-after header {exception}. Using default retry time: {DEFAULT_RETRY_SECONDS} seconds"
)
else:
self._logger.warning(
f"Rate Limited but Retry-After header is not found, using default retry time: {DEFAULT_RETRY_SECONDS} seconds"
)
self._logger.debug(f"Rate Limit reached: retry in {retry_seconds} seconds")
await self._sleeps.sleep(retry_seconds)
raise ThrottledError
elif exception.status == 404:
self._logger.error(f"Getting Not Found Error for url: {url}")
raise NotFound
elif exception.status == 500:
self._logger.error("Internal Server Error occurred")
raise InternalServerError
else:
raise
@retryable(
retries=RETRIES,
interval=RETRY_INTERVAL,
strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
skipped_exceptions=NotFound,
)
async def api_call(self, url):
"""Make a GET call for Atlassian API using the passed url with retry for the failed API calls.
Args:
url: Request URL to hit the get call
Raises:
exception: An instance of an exception class.
Yields:
response: Client response
"""
self._logger.debug(f"Making a GET call for url: {url}")
try:
async with self._get_session().get(
url=url,
ssl=self.ssl_ctx,
) as response:
yield response
except ServerDisconnectedError:
await self.close_session()
raise
except ClientResponseError as exception:
await self._handle_client_errors(url=url, exception=exception)
async def paginated_api_call(self, url_name, **url_kwargs):
"""Make a paginated API call for Confluence objects using the passed url_name.
Args:
url_name (str): URL Name to identify the API endpoint to hit
Yields:
response: JSON response.
"""
self._logger.info(
f"Started pagination for the API endpoint: {URLS[url_name]} to host: {self.host_url}"
)
url = os.path.join(self.host_url, URLS[url_name].format(**url_kwargs))
while True:
try:
async for response in self.api_call(
url=url,
):
json_response = await response.json()
links = json_response.get("_links")
yield json_response
if links.get("next") is None:
return
url = os.path.join(
self.host_url,
links.get("next")[1:],
)
except Exception as exception:
self._logger.warning(
f"Skipping data for type {url_name} from {url}. Exception: {exception}."
)
break
async def paginated_api_call_for_datacenter_syncrule(self, url_name, **url_kwargs):
"""Make a paginated API call for datacenter using the passed url_name.
Args:
url_name (str): URL Name to identify the API endpoint to hit
Yields:
response: JSON response.
"""
self._logger.info(
f"Started pagination for the API endpoint: {URLS[url_name]} to host: {self.host_url} with the parameters -> start: 0, limit: {LIMIT}"
)
while True:
url = os.path.join(self.host_url, URLS[url_name].format(**url_kwargs))
json_response = {}
try:
async for response in self.api_call(
url=url,
):
json_response = await response.json()
yield json_response
start = url_kwargs.get("start", 0)
start += LIMIT
url_kwargs["start"] = start
if len(json_response.get("results", [])) < LIMIT:
break
except Exception as exception:
self._logger.warning(
f"Skipping data for type {url_name} from {url}. Exception: {exception}."
)
break
async def search_by_query(self, query):
if self.data_source_type == CONFLUENCE_DATA_CENTER:
search_documents = self.paginated_api_call_for_datacenter_syncrule(
url_name=SEARCH_FOR_DATA_CENTER,
query=f"{query}&{SEARCH_QUERY}",
start=0,
)
else:
search_documents = self.paginated_api_call(
url_name=SEARCH,
query=f"{query}&{SEARCH_QUERY}",
)
async for response in search_documents:
for entity in response.get("results", []):
yield entity
async def fetch_spaces(self):
async for response in self.paginated_api_call(
url_name=SPACE,
api_query=SPACE_QUERY,
):
for space in response.get("results", []):
spaces = self.configuration.get("spaces", "")
if (spaces == [WILDCARD]) or (space.get("key", "") in spaces):
yield space
async def fetch_server_space_permission(self, url):
try:
async for permissions in self.api_call(
url=os.path.join(self.host_url, url),
):
permission = await permissions.json()
return permission
except ClientResponseError as exception:
self._logger.warning(
f"Something went wrong. Make sure you have installed Extender for running confluence datacenter/server DLS. Exception: {exception}."
)
return {}
async def fetch_page_blog_documents(self, api_query):
async for response in self.paginated_api_call(
url_name=CONTENT,
api_query=api_query,
):
attachment_count = 0
for document in response.get("results", []):
if document.get("children").get("attachment"):
attachment_count = (
document.get("children", {})
.get("attachment", {})
.get("size", 0)
)
if self.index_labels:
labels = await self.fetch_label(document["id"])
document["labels"] = labels
yield document, attachment_count
async def fetch_attachments(self, content_id):
async for response in self.paginated_api_call(
url_name=ATTACHMENT,
api_query=ATTACHMENT_QUERY,
id=content_id,
):
for attachment in response.get("results", []):
yield attachment
async def ping(self):
await anext(
self.api_call(
url=os.path.join(self.host_url, PING_URL),
)
)
async def fetch_confluence_server_users(self):
start_at = 0
if self.data_source_type == CONFLUENCE_DATA_CENTER:
limit = DATACENTER_USER_BATCH
key = "results"
url = urljoin(self.host_url, URLS[USERS_FOR_DATA_CENTER])
else:
limit = SERVER_USER_BATCH
key = "users"
url = urljoin(self.host_url, URLS[USERS_FOR_SERVER])
while True:
url_ = url.format(start=start_at, limit=limit)
async for users in self.api_call(url=url_):
response = await users.json()
if len(response.get(key)) == 0:
return
yield response.get(key)
start_at += limit
async def fetch_label(self, label_id):
async for label_data in self.api_call(
url=os.path.join(self.host_url, URLS[LABEL].format(id=label_id))
):
labels = await label_data.json()
return [label.get("name") for label in labels["results"]]
class ConfluenceDataSource(BaseDataSource):
"""Confluence"""
name = "Confluence"
service_type = "confluence"
advanced_rules_enabled = True
dls_enabled = True
incremental_sync_enabled = True
def __init__(self, configuration):
"""Setup the connection to Confluence
Args:
configuration (DataSourceConfiguration): Object of DataSourceConfiguration class.
"""
super().__init__(configuration=configuration)
self.spaces = self.configuration["spaces"]
self.concurrent_downloads = self.configuration["concurrent_downloads"]
self.confluence_client = ConfluenceClient(configuration=configuration)
self.atlassian_access_control = AtlassianAccessControl(
self, self.confluence_client
)
self.queue = MemQueue(maxsize=QUEUE_SIZE, maxmemsize=QUEUE_MEM_SIZE)
self.fetchers = ConcurrentTasks(max_concurrency=MAX_CONCURRENCY)
self.fetcher_count = 0
def _set_internal_logger(self):
self.confluence_client.set_logger(self._logger)
@classmethod
def get_default_configuration(cls):
"""Get the default configuration for Confluence
Returns:
dictionary: Default configuration.
"""
return {
"data_source": {
"display": "dropdown",
"label": "Confluence data source",
"options": [
{"label": "Confluence Cloud", "value": CONFLUENCE_CLOUD},
{"label": "Confluence Server", "value": CONFLUENCE_SERVER},
{
"label": "Confluence Data Center",
"value": CONFLUENCE_DATA_CENTER,
},
],
"order": 1,
"type": "str",
"value": CONFLUENCE_SERVER,
},
"username": {
"depends_on": [{"field": "data_source", "value": CONFLUENCE_SERVER}],
"label": "Confluence Server username",
"order": 2,
"type": "str",
},
"password": {
"depends_on": [{"field": "data_source", "value": CONFLUENCE_SERVER}],
"label": "Confluence Server password",
"sensitive": True,
"order": 3,
"type": "str",
},
"data_center_username": {
"depends_on": [
{"field": "data_source", "value": CONFLUENCE_DATA_CENTER}
],
"label": "Confluence Data Center username",
"order": 4,
"type": "str",
},
"data_center_password": {
"depends_on": [
{"field": "data_source", "value": CONFLUENCE_DATA_CENTER}
],
"label": "Confluence Data Center password",
"sensitive": True,
"order": 5,
"type": "str",
},
"account_email": {
"depends_on": [{"field": "data_source", "value": CONFLUENCE_CLOUD}],
"label": "Confluence Cloud account email",
"order": 6,
"type": "str",
},
"api_token": {
"depends_on": [{"field": "data_source", "value": CONFLUENCE_CLOUD}],
"label": "Confluence Cloud API token",
"sensitive": True,
"order": 7,
"type": "str",
},
"confluence_url": {
"label": "Confluence URL",
"order": 8,
"type": "str",
},
"spaces": {
"display": "textarea",
"label": "Confluence space keys",
"order": 9,
"tooltip": "This configurable field is ignored when Advanced Sync Rules are used.",
"type": "list",
},
"index_labels": {
"display": "toggle",
"label": "Enable indexing labels",
"order": 10,
"tooltip": "Enabling this will increase the amount of network calls to the source, and may decrease performance",
"type": "bool",
"value": False,
},
"ssl_enabled": {
"display": "toggle",
"label": "Enable SSL",
"order": 11,
"type": "bool",
"value": False,
},
"ssl_ca": {
"depends_on": [{"field": "ssl_enabled", "value": True}],
"label": "SSL certificate",
"order": 12,
"type": "str",
},
"retry_count": {
"default_value": 3,
"display": "numeric",
"label": "Retries per request",
"order": 13,
"required": False,
"type": "int",
"ui_restrictions": ["advanced"],
},
"concurrent_downloads": {
"default_value": MAX_CONCURRENT_DOWNLOADS,
"display": "numeric",
"label": "Maximum concurrent downloads",
"order": 14,
"required": False,
"type": "int",
"ui_restrictions": ["advanced"],
"validations": [
{"type": "less_than", "constraint": MAX_CONCURRENT_DOWNLOADS + 1}
],
},
"use_document_level_security": {
"display": "toggle",
"label": "Enable document level security",
"order": 15,
"tooltip": "Document level security ensures identities and permissions set in confluence are maintained in Elasticsearch. This enables you to restrict and personalize read-access users have to documents in this index. Access control syncs ensure this metadata is kept up to date in your Elasticsearch documents.",
"type": "bool",
"value": False,
},
"use_text_extraction_service": {
"display": "toggle",
"label": "Use text extraction service",
"order": 16,
"tooltip": "Requires a separate deployment of the Elastic Text Extraction Service. Requires that pipeline settings disable text extraction.",
"type": "bool",
"ui_restrictions": ["advanced"],
"value": False,
},
}
def _dls_enabled(self):
"""Check if document level security is enabled. This method checks whether document level security (DLS) is enabled based on the provided configuration.
Returns:
bool: True if document level security is enabled, False otherwise.
"""
if self._features is None:
return False
if not self._features.document_level_security_enabled():
return False
return self.configuration["use_document_level_security"]
def _decorate_with_access_control(self, document, access_control):
if self._dls_enabled():
document[ACCESS_CONTROL] = list(
set(document.get(ACCESS_CONTROL, []) + access_control)
)
return document
async def user_access_control_confluence_server(self, user):
"""Generate a user access control document for confluence server.
This method generates a user access control document based on the provided user information.
The document includes the user's account ID, prefixed account ID, and prefixed account name
The access control list is then constructed using these values.
Args:
user (dict): A dictionary containing user information, such as account ID, display name and email.
Returns:
dict: A user access control document with the following structure:
{
"_id": <account_id>,
"identity": {
"account_id": <prefixed_account_id>,
"display_name": <prefixed_account_name>
"email_address": <prefixed_account_email>
},
"created_at": <iso_utc_timestamp>,
ACCESS_CONTROL: [<prefixed_account_id>, <prefixed_account_email>, <prefixed_account_name>]
}
"""
account_id = user.get("key") or user.get("name")
account_name = user.get("fullName")
email = user.get("email")
prefixed_account_email = prefix_account_email(email=email)
prefixed_account_id = prefix_account_id(account_id=account_id)
prefixed_account_name = prefix_account_name(account_name=account_name)
access_control = [
prefixed_account_id,
prefixed_account_email,
prefixed_account_name,
]
return {
"_id": account_id,
"identity": {
"account_id": prefixed_account_id,
"display_name": prefixed_account_name,
"email_address": prefixed_account_email,
},
"created_at": iso_utc(),
} | self.atlassian_access_control.access_control_query(
access_control=access_control
)
async def user_access_control_data_center(self, user):
"""Generate a user access control document for confluence data enter.
This method generates a user access control document based on the provided user information.
The document includes the user's account ID, prefixed account ID, and prefixed account name
The access control list is then constructed using these values.
Args:
user (dict): A dictionary containing user information, such as account ID, display name and email.
Returns:
dict: A user access control document with the following structure:
{
"_id": <account_id>,
"identity": {
"account_id": <prefixed_account_id>,
"username": <prefixed_account_name>,
"display_name: <prefixed_display_name>
},
"created_at": <iso_utc_timestamp>,
ACCESS_CONTROL: [<prefixed_account_id>, <prefixed_account_name>, <prefixed_display_name>]
}
"""
account_id = user.get("userKey")
account_name = user.get("username")
display_name = user.get("displayName")
prefixed_account_id = prefix_account_id(account_id=account_id)
prefixed_account_name = prefix_account_name(account_name=account_name)
prefixed_display_name = prefix_account_name(account_name=display_name)
access_control = [
prefixed_account_id,
prefixed_account_name,
prefixed_display_name,
]
return {
"_id": account_id,
"identity": {
"account_id": prefixed_account_id,
"username": prefixed_account_name,
"display_name": prefixed_display_name,
},
"created_at": iso_utc(),
} | self.atlassian_access_control.access_control_query(
access_control=access_control
)
async def get_user_for_server(self):
async for users in self.confluence_client.fetch_confluence_server_users():
for user_info in users:
if self.confluence_client.data_source_type == CONFLUENCE_DATA_CENTER:
yield await self.user_access_control_data_center(user=user_info)
else:
yield await self.user_access_control_confluence_server(
user=user_info
)
async def get_user(self):
url = os.path.join(self.configuration["confluence_url"], URLS[USER])
async for users in self.atlassian_access_control.fetch_all_users(url=url):
active_atlassian_users = filter(
self.atlassian_access_control.is_active_atlassian_user, users
)
tasks = [
anext(
self.atlassian_access_control.fetch_user(
url=f"{user_info.get('self')}&{USER_QUERY}"
)
)
for user_info in active_atlassian_users
]
user_results = await asyncio.gather(*tasks)
for user in user_results:
yield await self.atlassian_access_control.user_access_control_doc(
user=user
)
async def get_access_control(self):
"""Get access control documents for active Atlassian users.
This method fetches access control documents for active Atlassian users when document level security (DLS)
is enabled.
If DLS is enabled, the method fetches all users from the Confluence API, filters out active Atlassian users,
and fetches additional information for each active user using the fetch_user method. After gathering the user information,
it generates an access control document for each user using the user_access_control_doc method and yields the results.
Yields:
dict: An access control document for each active Atlassian user.
"""
if not self._dls_enabled():
self._logger.warning("DLS is not enabled. Skipping")
return
self._logger.info("Fetching all users for Access Control sync")
if self.confluence_client.data_source_type == CONFLUENCE_CLOUD:
users = self.get_user()
else:
users = self.get_user_for_server()
async for user in users:
yield user
def _get_access_control_from_permission(self, permissions, target_type):
if not self._dls_enabled():
return []
access_control = set()
for permission in permissions:
permission_operation = permission.get("operation", {})
if permission_operation.get("targetType") != target_type and (
permission_operation.get("targetType") != SPACE
and permission_operation.get("operation") != "read"
):
continue
access_control = access_control.union(
self._extract_identities(response=permission.get("subjects", {}))
)
return access_control
def _extract_identities(self, response):
if not self._dls_enabled():
return set()
identities = set()
user_results = response.get("user", {}).get("results", [])
group_results = response.get("group", {}).get("results", [])
for item in user_results + group_results:
item_type = item.get("type")
if item_type == "known" and item.get("accountType") == "atlassian":
identities.add(prefix_account_id(account_id=item.get("accountId", "")))
elif item_type == "group":
identities.add(prefix_group_id(group_id=item.get("id", "")))
return identities
def _extract_identities_for_datacenter(self, response):
if not self._dls_enabled():
return set()
identities = set()
user_results = response.get("user", {}).get("results", [])
group_results = response.get("group", {}).get("results", [])
for item in user_results + group_results:
item_type = item.get("type")
if item_type == "known":
identities.add(prefix_user(item.get("username")))
elif item_type == "group":
identities.add(prefix_group(item.get("name", "")))
return identities
async def close(self):
"""Closes unclosed client session"""
await self.confluence_client.close_session()
def advanced_rules_validators(self):
return [AtlassianAdvancedRulesValidator(self)]
def tweak_bulk_options(self, options):
"""Tweak bulk options as per concurrent downloads support by Confluence
Args:
options (dictionary): Config bulker options
"""
options["concurrent_downloads"] = self.concurrent_downloads
async def validate_config(self):
"""Validates whether user input is empty or not for configuration fields.
Also validate, if user configured spaces are available in Confluence.
Raises:
Exception: Configured keys can't be empty
"""
await super().validate_config()
await self._remote_validation()
async def _remote_validation(self):
await self.confluence_client.ping()
if self.spaces == [WILDCARD]:
return
space_keys = []
async for response in self.confluence_client.paginated_api_call(
url_name=SPACE, api_query=SPACE_QUERY
):
spaces = response.get("results", [])
space_keys.extend([space.get("key", "") for space in spaces])
if unavailable_spaces := set(self.spaces) - set(space_keys):
msg = f"Spaces '{', '.join(unavailable_spaces)}' are not available. Available spaces are: '{', '.join(space_keys)}'"
raise ConfigurableFieldValueError(msg)
async def ping(self):
"""Verify the connection with Confluence"""
try:
await self.confluence_client.ping()
self._logger.info("Successfully connected to Confluence")
except Exception:
self._logger.exception("Error while connecting to Confluence")
raise
def get_permission(self, permission):
permissions = set()
if permission.get("users"):
for user in permission.get("users"):
permissions.add(prefix_user(user))
if permission.get("groups"):
for group in permission.get("groups"):
permissions.add(prefix_group(group))
return permissions
async def fetch_server_space_permission(self, space_key):
if not self._dls_enabled():
return {}
url = URLS[SPACE_PERMISSION].format(space_key=space_key)
self._logger.debug(
f"Fetching permissions for space '{space_key} from Confluence server'"
)
return await self.confluence_client.fetch_server_space_permission(url=url)
async def fetch_documents(self, api_query):
"""Get pages and blog posts with the help of REST APIs
Args:
api_query (str): Query parameter for the API call
Yields:
Dictionary: Page or blog post to be indexed
Integer: Number of attachments in a page/blogpost
List: List of permissions attached to document
Dictionary: Dictionary of restrictions attached to document
"""
async for document, attachment_count in self.confluence_client.fetch_page_blog_documents(
api_query=api_query,
):
document_url = os.path.join(
self.confluence_client.host_url,
document.get("_links", {}).get("webui", "")[1:],
)
ancestor_title = [
{"title": ancestor["title"]}
for ancestor in document.get("ancestors", [])
]
doc = {
"_id": str(document["id"]),
"type": document["type"],
"_timestamp": document["history"]["lastUpdated"]["when"],
"title": document.get("title"),
"ancestors": ancestor_title,
"space": document["space"]["name"],
"body": document["body"]["storage"]["value"],
"url": document_url,
}
if self.confluence_client.index_labels:
doc["labels"] = document["labels"]
yield doc, attachment_count, document.get("space", {}).get(
"key"
), document.get("space", {}).get("permissions", []), document.get(
"restrictions", {}
).get(
"read", {}
).get(
"restrictions", {}
)
async def fetch_attachments(
self, content_id, parent_name, parent_space, parent_type
):
"""Fetches all the attachments present in the given content (pages and blog posts)
Args:
content_id (str): Unique identifier of Confluence content (pages and blogposts)
parent_name (str): Page/Blog post where the attachment is present
parent_space (str): Confluence Space where the attachment is present
parent_type (str): Type of the parent (Blog post or Page)
Yields:
Dictionary: Confluence attachment on the given page or blog post
String: Download link to get the content of the attachment
"""
self._logger.info(
f"Fetching attachments for '{parent_name}' from '{parent_space}' space"
)
async for attachment in self.confluence_client.fetch_attachments(
content_id=content_id,
):
attachment_url = os.path.join(
self.confluence_client.host_url,
attachment.get("_links", {}).get("webui", "")[1:],
)
yield {
"type": attachment.get("type"),
"title": attachment.get("title"),
"_id": attachment.get("id"),
"space": parent_space,
parent_type: parent_name,
"_timestamp": attachment.get("version", {}).get("when", iso_utc()),
"size": attachment.get("extensions", {}).get("fileSize", 0),
"url": attachment_url,
}, attachment.get("_links", {}).get("download")
async def search_by_query(self, query):
async for entity in self.confluence_client.search_by_query(query=query):
# entity can be space or content
entity_details = entity.get(SPACE) or entity.get(CONTENT)
if (
entity_details.get("type", "") == "attachment"
and entity_details.get("container", {}).get("title") is None
):
continue
document = {
"_id": entity_details.get("id"),
"title": entity.get("title"),
"_timestamp": entity.get("lastModified"),
"body": entity.get("excerpt"),
"type": entity.get("entityType"),
"url": os.path.join(
self.confluence_client.host_url, entity.get("url")[1:]
),
}
download_url = None
if document.get("type", "") == "content":
document.update(
{
"type": entity_details.get("type"),
"space": entity_details.get("space", {}).get("name"),
}
)
if document.get("type", "") == "attachment":
container_type = entity_details.get("container", {}).get("type")
container_title = entity_details.get("container", {}).get("title")
file_size = entity_details.get("extensions", {}).get("fileSize")
document.update(
{"size": file_size, container_type: container_title}
)
# Removing body as attachment will be downloaded lazily
document.pop("body")
download_url = entity_details.get("_links", {}).get("download")
yield document, download_url
async def download_attachment(self, url, attachment, timestamp=None, doit=False):
"""Downloads the content of the given attachment in chunks using REST API call
Args:
url (str): url endpoint to download the attachment content
attachment (dict): Dictionary containing details of the attachment
timestamp (timestamp, optional): Timestamp of Confluence last modified. Defaults to None.
doit (boolean, optional): Boolean value for whether to get content or not. Defaults to False.
Returns:
Dictionary: Document of the attachment to be indexed.