Merge pull request #2 from C4T-BuT-S4D/falamous/crypter

add crypter service

Author: pomo-mondreganto

checkers/crypter/checker.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+
+import random
+import secrets
+import sys
+import uuid
+import json
+
+import grpc
+from checklib import *
+
+from crypter_lib import CheckMachine
+
+
+class Checker(BaseChecker):
+    vulns: int = 1
+    timeout: int = 15
+    uses_attack_data: bool = True
+
+    def __init__(self, *args, **kwargs):
+        super(Checker, self).__init__(*args, **kwargs)
+        self.c = CheckMachine(self)
+
+    def action(self, action, *args, **kwargs):
+        try:
+            super(Checker, self).action(action, *args, **kwargs)
+        except self.get_check_finished_exception():
+            raise
+        except grpc.RpcError as e:
+            if e.code() == grpc.StatusCode.UNAVAILABLE:
+                self.cquit(Status.DOWN, "unavailable", f"grpc error: {e}")
+            else:
+                self.cquit(Status.MUMBLE, f"grpc error: {e.code()}", f"grpc error: {e}")
+        except ConnectionRefusedError:
+            self.cquit(Status.DOWN, "Connection refused", "Connection refused")
+
+    def check(self):
+        with self.c.connect() as channel, self.c.handle_grpc_error(status=Status.MUMBLE):
+            stub = self.c.get_stub(channel)
+
+            username = rnd_username()
+            user = self.c.Register(stub, username)
+
+            from_username = rnd_username()
+            from_user = self.c.Register(stub, from_username)
+
+            data = rnd_string(32)
+            message_id = self.c.SendMessage(stub, username, from_user.token, data)
+            n = self.c.GetUserPublicKey(stub, username)
+
+            message = self.c.GetMessage(stub, message_id)
+
+            plaintext = self.c.int_to_bytes(self.c.decrypt(user.n, user.lamba, message.encrypted))
+
+            self.assert_eq(message.username, username, "incorrect username", Status.MUMBLE)
+            self.assert_eq(message.from_username, from_username, "incorrect from_username", Status.MUMBLE)
+            self.assert_eq(plaintext, data.encode(), "incorrect flag", Status.MUMBLE)
+
+            self.cquit(Status.OK)
+
+
+    def put(self, flag_id: str, flag: str, vuln: str):
+        with self.c.connect() as channel, self.c.handle_grpc_error(status=Status.MUMBLE):
+            stub = self.c.get_stub(channel)
+
+            username = rnd_username()
+            user = self.c.Register(stub, username)
+
+            from_username = rnd_username()
+            from_user = self.c.Register(stub, from_username)
+
+            message_id = self.c.SendMessage(stub, username, from_user.token, flag)
+            self.cquit(Status.OK, 
+                       json.dumps({
+                           "message": message_id,
+                       }),
+                       json.dumps({   
+                           "username": username,
+                       "lambda": user.lamba,
+                       "from": from_username,
+                           "message": message_id,
+                       }),
+                       )
+
+
+    def get(self, flag_id: str, flag: str, vuln: str):
+        with self.c.connect() as channel, self.c.handle_grpc_error(status=Status.CORRUPT):
+            stub = self.c.get_stub(channel)
+
+            flag_data = json.loads(flag_id)
+
+            lamba = flag_data["lambda"]
+            username = flag_data["username"]
+            from_username = flag_data["from"]
+            message_id = flag_data["message"]
+
+            n = self.c.GetUserPublicKey(stub, username)
+
+            message = self.c.GetMessage(stub, message_id)
+
+            plaintext = self.c.int_to_bytes(self.c.decrypt(n, lamba, message.encrypted))
+
+            self.assert_eq(message.username, username, "incorrect username", Status.CORRUPT)
+            self.assert_eq(message.from_username, from_username, "incorrect from_username", Status.CORRUPT)
+            self.assert_eq(plaintext, flag.encode(), "incorrect flag", Status.CORRUPT)
+
+            self.cquit(Status.OK)
+
+
+
+if __name__ == "__main__":
+    c = Checker(sys.argv[2])
+
+    try:
+        c.action(sys.argv[1], *sys.argv[3:])
+    except c.get_check_finished_exception():
+        cquit(Status(c.status), c.public, c.private)

checkers/crypter/crypter_lib.py

@@ -0,0 +1,130 @@
+import random
+import string
+import uuid
+from typing import Callable, List, NamedTuple
+
+from checklib import *
+import grpc
+from contextlib import contextmanager
+from crypter_pb2 import (
+    RegisterRequest,
+    RegisterResponse,
+    GetMessageRequest,
+    GetMessageResponse,
+    SendMessageRequest,
+    SendMessageResponse,
+    GetUserPublicKeyRequest,
+    GetUserPublicKeyResponse,
+)
+from crypter_pb2_grpc import CrypterStub
+
+PORT = 2112
+
+class User(NamedTuple):
+    token: str
+    n: int
+    lamba: int
+
+class EncryptedMessage(NamedTuple):
+    username: string
+    from_username: string
+    encrypted: int
+
+class CheckMachine:
+    def __init__(self, c: BaseChecker):
+        self.c = c
+        self.addr = f"{self.c.host}:{PORT}"
+
+    def connect(self) -> grpc.Channel:
+        channel = grpc.insecure_channel(self.addr)
+        return channel
+
+    @staticmethod
+    def get_stub(channel): 
+        return CrypterStub(channel)
+
+    @staticmethod
+    def Register(
+             stub: CrypterStub,
+                username: string
+             ) -> User:
+        resp: RegisterResponse = stub.Register(
+            RegisterRequest(
+                username=username,
+            )
+        )
+        return User(
+            token=resp.token,
+            n=int(resp.n),
+            lamba=int(resp.lamba),
+        )
+
+    @staticmethod
+    def GetMessage(
+             stub: CrypterStub,
+            id: string,
+             ) -> EncryptedMessage:
+        resp: GetMessageResponse = stub.GetMessage(
+            GetMessageRequest(
+                id=id,
+            )
+        )
+        return EncryptedMessage(
+            username=resp.username,
+            from_username=resp.from_username,
+            encrypted=int(resp.encrypted),
+        )
+
+    @staticmethod
+    def SendMessage(
+             stub: CrypterStub,
+            username: string,
+            token: string,
+        message: string,
+             ) -> str:
+        resp: SendMessageResponse = stub.SendMessage(
+            SendMessageRequest(
+                username=username,
+                token=token,
+                message=message,
+            )
+        )
+        return resp.id
+
+    @staticmethod
+    def GetUserPublicKey(
+             stub: CrypterStub,
+            username: string,
+             ) -> str:
+        resp: GetUserPublicKeyResponse = stub.GetUserPublicKey(
+            GetUserPublicKeyRequest(
+                username=username
+            )
+        )
+        return int(resp.n)
+
+    @contextmanager
+    def handle_grpc_error(self, status=Status.MUMBLE):
+        try:
+            yield
+        except grpc.RpcError as e:
+            if e.code() == grpc.StatusCode.UNAVAILABLE:
+                raise
+            else:
+                self.c.cquit(status, f"grpc error: {e.code()}", f"grpc error: {e}")
+        except ValueError as e:
+            self.c.cquit(status, f"value error", f"value error: {e}")
+
+    @staticmethod
+    def fake_flag() -> str:
+        return (
+            "B" + rnd_string(31, alphabet=string.ascii_uppercase + string.digits) + "="
+        )
+
+    @staticmethod
+    def int_to_bytes(n: int) -> bytes:
+        return n.to_bytes((n.bit_length() + 7 )// 8, "big")
+
+    @staticmethod
+    def decrypt(n: int, lamba: int, e: int):
+        return (pow(e, lamba, n ** 2) - 1) // n * pow((pow(n + 1, lamba, n ** 2) - 1) // n, -1, n) % n

checkers/crypter/crypter_pb2.py

@@ -0,0 +1,78 @@
+from google.protobuf.internal import containers as _containers
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from typing import ClassVar as _ClassVar, Iterable as _Iterable, Optional as _Optional
+
+DESCRIPTOR: _descriptor.FileDescriptor
+
+class RegisterRequest(_message.Message):
+    __slots__ = ("username",)
+    USERNAME_FIELD_NUMBER: _ClassVar[int]
+    username: str
+    def __init__(self, username: _Optional[str] = ...) -> None: ...
+
+class RegisterResponse(_message.Message):
+    __slots__ = ("token", "n", "lamba")
+    TOKEN_FIELD_NUMBER: _ClassVar[int]
+    N_FIELD_NUMBER: _ClassVar[int]
+    LAMBA_FIELD_NUMBER: _ClassVar[int]
+    token: str
+    n: str
+    lamba: str
+    def __init__(self, token: _Optional[str] = ..., n: _Optional[str] = ..., lamba: _Optional[str] = ...) -> None: ...
+
+class SendMessageRequest(_message.Message):
+    __slots__ = ("token", "username", "message")
+    TOKEN_FIELD_NUMBER: _ClassVar[int]
+    USERNAME_FIELD_NUMBER: _ClassVar[int]
+    MESSAGE_FIELD_NUMBER: _ClassVar[int]
+    token: str
+    username: str
+    message: str
+    def __init__(self, token: _Optional[str] = ..., username: _Optional[str] = ..., message: _Optional[str] = ...) -> None: ...
+
+class SendMessageResponse(_message.Message):
+    __slots__ = ("id",)
+    ID_FIELD_NUMBER: _ClassVar[int]
+    id: str
+    def __init__(self, id: _Optional[str] = ...) -> None: ...
+
+class ListMessagesRequest(_message.Message):
+    __slots__ = ("token",)
+    TOKEN_FIELD_NUMBER: _ClassVar[int]
+    token: str
+    def __init__(self, token: _Optional[str] = ...) -> None: ...
+
+class ListMessagesResponse(_message.Message):
+    __slots__ = ("id",)
+    ID_FIELD_NUMBER: _ClassVar[int]
+    id: _containers.RepeatedScalarFieldContainer[str]
+    def __init__(self, id: _Optional[_Iterable[str]] = ...) -> None: ...
+
+class GetMessageRequest(_message.Message):
+    __slots__ = ("id",)
+    ID_FIELD_NUMBER: _ClassVar[int]
+    id: str
+    def __init__(self, id: _Optional[str] = ...) -> None: ...
+
+class GetMessageResponse(_message.Message):
+    __slots__ = ("username", "from_username", "encrypted")
+    USERNAME_FIELD_NUMBER: _ClassVar[int]
+    FROM_USERNAME_FIELD_NUMBER: _ClassVar[int]
+    ENCRYPTED_FIELD_NUMBER: _ClassVar[int]
+    username: str
+    from_username: str
+    encrypted: str
+    def __init__(self, username: _Optional[str] = ..., from_username: _Optional[str] = ..., encrypted: _Optional[str] = ...) -> None: ...
+
+class GetUserPublicKeyRequest(_message.Message):
+    __slots__ = ("username",)
+    USERNAME_FIELD_NUMBER: _ClassVar[int]
+    username: str
+    def __init__(self, username: _Optional[str] = ...) -> None: ...
+
+class GetUserPublicKeyResponse(_message.Message):
+    __slots__ = ("n",)
+    N_FIELD_NUMBER: _ClassVar[int]
+    n: str
+    def __init__(self, n: _Optional[str] = ...) -> None: ...