Repository: Chia-Network/chia-blockchain
File: chia/server/server.py (Python)

Code before:
    ca_cert: Path,
    ca_key: Path,
    cert_path: Path,
    key_path: Path,
    *,
    check_permissions: bool = True,
    log: logging.Logger | None = None,
) -> ssl.SSLContext:
    if check_permissions:
        verify_ssl_certs_and_keys([ca_cert, cert_path], [ca_key, key_path], log)

    ssl_context = ssl._create_unverified_context(purpose=ssl.Purpose.SERVER_AUTH, cafile=str(ca_cert))  # noqa: S323
    ssl_context.check_hostname = False
    ssl_context.load_cert_chain(certfile=str(cert_path), keyfile=str(key_path))
    ssl_context.verify_mode = ssl.CERT_REQUIRED
    return ssl_context


def calculate_node_id(cert_path: Path) -> bytes32:
    pem_cert = x509.load_pem_x509_certificate(cert_path.read_bytes(), default_backend())
    der_cert_bytes = pem_cert.public_bytes(encoding=serialization.Encoding.DER)
    der_cert = x509.load_der_x509_certificate(der_cert_bytes, default_backend())
    return bytes32(der_cert.fingerprint(hashes.SHA256()))


def get_inbound_limit_defaults() -> dict[NodeType, tuple[str, int]]:
    """
    Returns a map of NodeType to tuple of config key and default value for
    inbound limit.
    """
    inbound_limit_defaults = {
        NodeType.HARVESTER: ("max_inbound_harvester", 5),
        NodeType.FARMER: ("max_inbound_farmer", 10),
        NodeType.TIMELORD: ("max_inbound_timelord", 5),
        NodeType.INTRODUCER: ("max_inbound_introducer", 5),
        NodeType.WALLET: ("max_inbound_wallet", 20),
        NodeType.DATA_LAYER: ("max_inbound_data_layer", 5),
        NodeType.SOLVER: ("max_inbound_solver", 5),
    }
    assert set(NodeType) - {NodeType.FULL_NODE} == set(inbound_limit_defaults)
    return inbound_limit_defaults


@final
@dataclass
class ChiaServer:
    _port: int | None
    _local_type: NodeType
    _local_capabilities_for_handshake: list[tuple[uint16, str]]
    _ping_interval: int
    _network_id: str

Review comment (suggestion):
This is a special case of creating new user config parameters and setting default values to use default parameters for ones are missing from a legacy config file. Is there a way to do this in a more "config"-based file? Is seems a bit specific to have this config-override stuff at this level.

Code after:
    ca_cert: Path,
    ca_key: Path,
    cert_path: Path,
    key_path: Path,
    *,
    check_permissions: bool = True,
    log: logging.Logger | None = None,
) -> ssl.SSLContext:
    if check_permissions:
        verify_ssl_certs_and_keys([ca_cert, cert_path], [ca_key, key_path], log)

    ssl_context = ssl._create_unverified_context(purpose=ssl.Purpose.SERVER_AUTH, cafile=str(ca_cert))  # noqa: S323
    ssl_context.check_hostname = False
    ssl_context.load_cert_chain(certfile=str(cert_path), keyfile=str(key_path))
    ssl_context.verify_mode = ssl.CERT_REQUIRED
    return ssl_context


def calculate_node_id(cert_path: Path) -> bytes32:
    pem_cert = x509.load_pem_x509_certificate(cert_path.read_bytes(), default_backend())
    der_cert_bytes = pem_cert.public_bytes(encoding=serialization.Encoding.DER)
    der_cert = x509.load_der_x509_certificate(der_cert_bytes, default_backend())
    return bytes32(der_cert.fingerprint(hashes.SHA256()))


@final
@dataclass
class ChiaServer:
    _port: int | None
    _local_type: NodeType
    _local_capabilities_for_handshake: list[tuple[uint16, str]]
    _ping_interval: int
    _network_id: str
    _inbound_rate_limit_percent: int
    _outbound_rate_limit_percent: int
    api: ApiProtocol
    node: Any
    root_path: Path
    config: dict[str, Any]
    log: logging.Logger
    ssl_context: ssl.SSLContext
    ssl_client_context: ssl.SSLContext
    node_id: bytes32
    exempt_peer_networks: list[IPv4Network | IPv6Network]
    stub_metadata_for_type: dict[NodeType, ApiMetadata]
    all_connections: dict[bytes32, WSChiaConnection] = field(default_factory=dict)
    on_connect: ConnectionCallback | None = None
    shut_down_event: asyncio.Event = field(default_factory=asyncio.Event)
    introducer_peers: IntroducerPeers | None = None
    gc_task: asyncio.Task[None] | None = None
    webserver: WebServer | None = None