feat(hosts): Add pinnedPeerCertSha256 support

Author: ImMohammad20000

app/core/hosts.py

@@ -69,6 +69,7 @@ async def _prepare_subscription_inbound_data(
     if tls_value is None:
         tls_value = inbound_config.get("tls", "none")
 
+    pinnedPeerCertSha256 = host.pinnedPeerCertSha256
     alpn_list = [alpn.value for alpn in host.alpn] if host.alpn else inbound_config.get("alpn", [])
     fp = host.fingerprint.value if host.fingerprint.value != "none" else inbound_config.get("fp")
     fp = fp or ("chrome" if tls_value == "reality" else "")
@@ -81,6 +82,7 @@ async def _prepare_subscription_inbound_data(
         sni=sni_list,
         fingerprint=fp,
         allowinsecure=ais,
+        pinnedPeerCertSha256=pinnedPeerCertSha256,
         alpn_list=alpn_list,
         ech_config_list=host.ech_config_list,
         reality_public_key=reality_pbk,

app/db/migrations/versions/5213b80a795c_add_pinnedpeercertsha256_to_hosts.py

@@ -0,0 +1,37 @@
+"""add pinnedPeerCertSha256 to hosts
+
+Revision ID: 5213b80a795c
+Revises: 7b5e1623c19a
+Create Date: 2026-02-11 22:34:07.678777
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '5213b80a795c'
+down_revision = '7b5e1623c19a'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('ix_apscheduler_jobs_next_run_time'), table_name='apscheduler_jobs')
+    op.drop_table('apscheduler_jobs')
+    op.add_column('hosts', sa.Column('pinnedPeerCertSha256', sa.String(length=128), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('hosts', 'pinnedPeerCertSha256')
+    op.create_table('apscheduler_jobs',
+    sa.Column('id', sa.VARCHAR(length=191), nullable=False),
+    sa.Column('next_run_time', sa.FLOAT(), nullable=True),
+    sa.Column('job_state', sa.BLOB(), nullable=False),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_apscheduler_jobs_next_run_time'), 'apscheduler_jobs', ['next_run_time'], unique=False)
+    # ### end Alembic commands ###

app/db/models.py

@@ -450,6 +450,7 @@ class ProxyHost(Base):
     )
     ech_config_list: Mapped[Optional[str]] = mapped_column(String(512), default=None)
     vless_route: Mapped[Optional[str]] = mapped_column(String(4), default=None)
+    pinnedPeerCertSha256: Mapped[Optional[str]] = mapped_column(String(128), default=None)
 
 
 class System(Base):

app/models/host.py

@@ -238,6 +238,7 @@ class BaseHost(BaseModel):
     priority: int
     status: set[UserStatus] | None = Field(default_factory=set)
     ech_config_list: str | None = Field(default=None)
+    pinnedPeerCertSha256:str| None = Field(default=None)
 
     model_config = ConfigDict(from_attributes=True)
 

app/models/subscription.py

@@ -19,6 +19,7 @@ class TLSConfig(BaseModel):
     allowinsecure: bool = Field(False)
     alpn_list: list[str] = Field(default_factory=list)
     ech_config_list: str | None = Field(None)
+    pinnedPeerCertSha256: str | None = Field(default=None)
 
     # Reality specific
     reality_public_key: str = Field("")

app/subscription/links.py

@@ -173,6 +173,7 @@ def _apply_tls_settings(self, payload: dict, tls_config: TLSConfig, fragment_set
         sni = tls_config.sni if isinstance(tls_config.sni, str) else ""
         payload["sni"] = sni
         payload["fp"] = tls_config.fingerprint
+        payload["pcs"] = tls_config.pinnedPeerCertSha256
 
         # Use pre-formatted alpn for links (comma-separated string)
         if tls_config.alpn_links:

app/subscription/singbox.py

@@ -136,15 +136,13 @@ def _transport_ws(self, config: WebSocketTransportConfig, path: str) -> dict:
 
     def _transport_grpc(self, config: GRPCTransportConfig, path: str) -> dict:
         """Handle GRPC transport - only gets GRPC config"""
-        return self._normalize_and_remove_none_values(
-            {
-                "type": "grpc",
-                "service_name": path,
-                "idle_timeout": f"{config.idle_timeout}s" if config.idle_timeout else "15s",
-                "ping_timeout": f"{config.health_check_timeout}s" if config.health_check_timeout else "15s",
-                "permit_without_stream": config.permit_without_stream,
-            }
-        )
+        return self._normalize_and_remove_none_values({
+            "type": "grpc",
+            "service_name": path,
+            "idle_timeout": f"{config.idle_timeout}s" if config.idle_timeout else "15s",
+            "ping_timeout": f"{config.health_check_timeout}s" if config.health_check_timeout else "15s",
+            "permit_without_stream": config.permit_without_stream,
+        })
 
     def _transport_httpupgrade(self, config: WebSocketTransportConfig, path: str) -> dict:
         """Handle HTTPUpgrade transport - only gets WS config (similar to WS)"""
@@ -192,6 +190,9 @@ def _apply_tls(self, tls_config: TLSConfig, fragment_settings: dict | None = Non
             if isinstance(tls_config.sni, str)
             else (tls_config.sni[0] if tls_config.sni else None),
             "insecure": tls_config.allowinsecure,
+            "certificate_public_key_sha256": [tls_config.pinnedPeerCertSha256]
+            if tls_config.pinnedPeerCertSha256
+            else [],
             "utls": {
                 "enabled": bool(tls_config.fingerprint) or tls_config.tls == "reality",
                 "fingerprint": tls_config.fingerprint,

app/subscription/xray.py

@@ -261,27 +261,23 @@ def _transport_quic(self, config: QUICTransportConfig, path: str) -> dict:
         """Handle QUIC transport - only gets QUIC config"""
         host = config.host if isinstance(config.host, str) else (config.host[0] if config.host else "")
 
-        return self._normalize_and_remove_none_values(
-            {
-                "security": host,
-                "header": {"type": config.header_type},
-                "key": path,
-            }
-        )
+        return self._normalize_and_remove_none_values({
+            "security": host,
+            "header": {"type": config.header_type},
+            "key": path,
+        })
 
     def _transport_kcp(self, config: KCPTransportConfig, path: str) -> dict:
         """Handle KCP transport - only gets KCP config"""
-        return self._normalize_and_remove_none_values(
-            {
-                "mtu": config.mtu if config.mtu is not None else 1350,
-                "tti": config.tti if config.tti is not None else 50,
-                "uplinkCapacity": config.uplink_capacity if config.uplink_capacity is not None else 5,
-                "downlinkCapacity": config.downlink_capacity if config.downlink_capacity is not None else 20,
-                "congestion": config.congestion,
-                "readBufferSize": config.read_buffer_size if config.read_buffer_size is not None else 2,
-                "writeBufferSize": config.write_buffer_size if config.write_buffer_size is not None else 2,
-            }
-        )
+        return self._normalize_and_remove_none_values({
+            "mtu": config.mtu if config.mtu is not None else 1350,
+            "tti": config.tti if config.tti is not None else 50,
+            "uplinkCapacity": config.uplink_capacity if config.uplink_capacity is not None else 5,
+            "downlinkCapacity": config.downlink_capacity if config.downlink_capacity is not None else 20,
+            "congestion": config.congestion,
+            "readBufferSize": config.read_buffer_size if config.read_buffer_size is not None else 2,
+            "writeBufferSize": config.write_buffer_size if config.write_buffer_size is not None else 2,
+        })
 
     def _apply_transport(self, network: str, inbound: SubscriptionInboundData, path: str) -> dict | None:
         """Apply transport settings using registry pattern"""
@@ -297,24 +293,23 @@ def _apply_tls(self, tls_config: TLSConfig, security: str) -> dict:
         sni = tls_config.sni if isinstance(tls_config.sni, str) else (tls_config.sni[0] if tls_config.sni else None)
 
         if security == "reality":
-            return self._normalize_and_remove_none_values(
-                {
-                    "serverName": sni,
-                    "fingerprint": tls_config.fingerprint,
-                    "show": False,
-                    "publicKey": tls_config.reality_public_key,
-                    "shortId": tls_config.reality_short_id,
-                    "spiderX": tls_config.reality_spx,
-                    "mldsa65Verify": tls_config.mldsa65_verify,
-                }
-            )
+            return self._normalize_and_remove_none_values({
+                "serverName": sni,
+                "fingerprint": tls_config.fingerprint,
+                "show": False,
+                "publicKey": tls_config.reality_public_key,
+                "shortId": tls_config.reality_short_id,
+                "spiderX": tls_config.reality_spx,
+                "mldsa65Verify": tls_config.mldsa65_verify,
+            })
         else:  # tls
             config = {
                 "serverName": sni,
                 "allowInsecure": tls_config.allowinsecure,
                 "show": False,
                 "fingerprint": tls_config.fingerprint,
                 "echConfigList": tls_config.ech_config_list,
+                "pinnedPeerCertSha256": tls_config.pinnedPeerCertSha256,
             }
             if tls_config.alpn_list:
                 config["alpn"] = tls_config.alpn_list  # Use list for xray
@@ -371,13 +366,11 @@ def _download_config(self, download_settings: SubscriptionInboundData, link_form
             sockopt=sockopt,
         )
 
-        return self._normalize_and_remove_none_values(
-            {
-                "address": download_settings.address,
-                "port": self._select_port(download_settings.port),
-                **stream_settings,
-            }
-        )
+        return self._normalize_and_remove_none_values({
+            "address": download_settings.address,
+            "port": self._select_port(download_settings.port),
+            **stream_settings,
+        })
 
     # ========== Protocol Builders (Registry Methods) ==========