Get blocked collection via commandline

blocking.py

@@ -37,6 +37,8 @@ from utils import has_actor
 from utils import text_in_file
 from conversation import mute_conversation
 from conversation import unmute_conversation
+from auth import create_basic_auth_header
+from session import get_json
 
 
 def get_global_block_reason(search_text: str,
@@ -1640,3 +1642,39 @@ def export_blocking_file(base_dir: str, nickname: str, domain: str) -> str:
             blocked_domain + ',suspend,false,false,"' + \
             reason_str + '",false\n'
     return blocks_str
+
+
+def get_blocks_via_server(session, nickname: str, password: str,
+                          domain: str, port: int,
+                          http_prefix: str, debug: bool,
+                          signing_priv_key_pem: str) -> {}:
+    """Returns the blocked collection for shared items via c2s
+    https://codeberg.org/fediverse/fep/src/branch/main/fep/c648/fep-c648.md
+    """
+    if not session:
+        print('WARN: No session for get_blocks_via_server')
+        return 6
+
+    auth_header = create_basic_auth_header(nickname, password)
+
+    headers = {
+        'host': domain,
+        'Content-type': 'application/json',
+        'Authorization': auth_header,
+        'Accept': 'application/json'
+    }
+    domain_full = get_full_domain(domain, port)
+    url = local_actor_url(http_prefix, nickname, domain_full) + '/blocked'
+    if debug:
+        print('Blocked collection request to: ' + url)
+    blocked_json = get_json(signing_priv_key_pem, session, url, headers, None,
+                            debug, __version__, http_prefix, None)
+    if not blocked_json:
+        if debug:
+            print('DEBUG: GET blocked collection failed for c2s to ' + url)
+#        return 5
+
+    if debug:
+        print('DEBUG: c2s GET blocked collection success')
+
+    return blocked_json

epicyon.py

@@ -114,6 +114,7 @@ from happening import dav_month_via_server
 from happening import dav_day_via_server
 from content import import_emoji
 from relationships import get_moved_accounts
+from blocking import get_blocks_via_server
 
 
 def str2bool(value_str) -> bool:
@@ -413,6 +414,12 @@ def _command_options() -> None:
                         const=True, default=False,
                         help="Get the following list. Use nickname and " +
                         "domain options to specify the account")
+    parser.add_argument("--blocked",
+                        dest='blocked',
+                        type=str2bool, nargs='?',
+                        const=True, default=False,
+                        help="Get the blocked collection. Use nickname and " +
+                        "domain options to specify the account")
     parser.add_argument("--followersList",
                         dest='followersList',
                         type=str2bool, nargs='?',
@@ -2484,6 +2491,40 @@ def _command_options() -> None:
             pprint(following_json)
         sys.exit()
 
+    if argb.blocked:
+        # blocked collection
+        if not argb.nickname:
+            print('Please specify the nickname for the account ' +
+                  'with --nickname')
+            sys.exit()
+        if not argb.password:
+            argb.password = getpass.getpass('Password: ')
+            if not argb.password:
+                print('Specify a password with the --password option')
+                sys.exit()
+        argb.password = remove_eol(argb.password)
+
+        session = create_session(proxy_type)
+        person_cache = {}
+        cached_webfingers = {}
+        blocked_http_prefix = http_prefix
+        if not domain:
+            domain = get_config_param(base_dir, 'domain')
+        signing_priv_key_pem = None
+        if argb.secure_mode:
+            signing_priv_key_pem = get_instance_actor_key(base_dir, domain)
+
+        blocked_json = \
+            get_blocks_via_server(session,
+                                  argb.nickname, argb.password,
+                                  domain, port,
+                                  blocked_http_prefix, argb.pageNumber,
+                                  debug, __version__,
+                                  signing_priv_key_pem)
+        if blocked_json:
+            pprint(blocked_json)
+        sys.exit()
+
     if argb.followersList:
         # following list via c2s protocol
         if not argb.nickname:

person.py

@@ -806,7 +806,8 @@ def person_upgrade_actor(base_dir: str, person_json: {},
 
     if person_json.get('endpoints'):
         if not person_json['endpoints'].get('blocked'):
-            person_json['endpoints']['blocked'] = person_json['id'] + '/blocked'
+            person_json['endpoints']['blocked'] = \
+                person_json['id'] + '/blocked'
             update_actor = True
         if not person_json['endpoints'].get('offers'):
             person_json['endpoints']['offers'] = person_json['id'] + '/offers'