Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Add type hints to get_domain_from_id and get_localpart_from_id. #10385

Merged
merged 3 commits into from
Jul 13, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/10385.misc
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Add type hints to `get_{domain,localpart}_from_id`.
96 changes: 72 additions & 24 deletions synapse/federation/transport/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -1095,7 +1095,9 @@ async def on_GET(
query: Dict[bytes, List[bytes]],
group_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1110,7 +1112,9 @@ async def on_POST(
query: Dict[bytes, List[bytes]],
group_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1131,7 +1135,9 @@ async def on_GET(
query: Dict[bytes, List[bytes]],
group_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1152,7 +1158,9 @@ async def on_GET(
query: Dict[bytes, List[bytes]],
group_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1174,7 +1182,9 @@ async def on_POST(
group_id: str,
room_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1192,7 +1202,9 @@ async def on_DELETE(
group_id: str,
room_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand Down Expand Up @@ -1220,7 +1232,9 @@ async def on_POST(
room_id: str,
config_key: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1243,7 +1257,9 @@ async def on_GET(
query: Dict[bytes, List[bytes]],
group_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1264,7 +1280,9 @@ async def on_GET(
query: Dict[bytes, List[bytes]],
group_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1288,7 +1306,9 @@ async def on_POST(
group_id: str,
user_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand Down Expand Up @@ -1354,7 +1374,9 @@ async def on_POST(
group_id: str,
user_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand Down Expand Up @@ -1487,7 +1509,9 @@ async def on_POST(
category_id: str,
room_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand Down Expand Up @@ -1523,7 +1547,9 @@ async def on_DELETE(
category_id: str,
room_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1549,7 +1575,9 @@ async def on_GET(
query: Dict[bytes, List[bytes]],
group_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1571,7 +1599,9 @@ async def on_GET(
group_id: str,
category_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1589,7 +1619,9 @@ async def on_POST(
group_id: str,
category_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand Down Expand Up @@ -1618,7 +1650,9 @@ async def on_DELETE(
group_id: str,
category_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1644,7 +1678,9 @@ async def on_GET(
query: Dict[bytes, List[bytes]],
group_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1666,7 +1702,9 @@ async def on_GET(
group_id: str,
role_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand All @@ -1682,7 +1720,9 @@ async def on_POST(
group_id: str,
role_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand Down Expand Up @@ -1713,7 +1753,9 @@ async def on_DELETE(
group_id: str,
role_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand Down Expand Up @@ -1750,7 +1792,9 @@ async def on_POST(
role_id: str,
user_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand Down Expand Up @@ -1784,7 +1828,9 @@ async def on_DELETE(
role_id: str,
user_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand Down Expand Up @@ -1825,7 +1871,9 @@ async def on_PUT(
query: Dict[bytes, List[bytes]],
group_id: str,
) -> Tuple[int, JsonDict]:
requester_user_id = parse_string_from_args(query, "requester_user_id")
requester_user_id = parse_string_from_args(
query, "requester_user_id", required=True
)
if get_domain_from_id(requester_user_id) != origin:
raise SynapseError(403, "requester_user_id doesn't match origin")

Expand Down
4 changes: 2 additions & 2 deletions synapse/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,14 +182,14 @@ def create_requester(
)


def get_domain_from_id(string):
def get_domain_from_id(string: str) -> str:
idx = string.find(":")
if idx == -1:
raise SynapseError(400, "Invalid ID: %r" % (string,))
return string[idx + 1 :]


def get_localpart_from_id(string):
def get_localpart_from_id(string: str) -> str:
idx = string.find(":")
if idx == -1:
raise SynapseError(400, "Invalid ID: %r" % (string,))
Expand Down