protocols between controller and workers

admin/.env

@@ -0,0 +1,2 @@
+SERVER_HOST=localhost
+SERVER_PORT=50051

admin/README.md

@@ -0,0 +1,26 @@
+# Admin Client
+
+## Установка зависимостей
+```bash
+pip install -r requirements.txt
+```
+
+## Генерация proto-файлов
+```bash
+python -m grpc_tools.protoc \
+    --python_out=. \
+    --grpc_python_out=. \
+    -I . \
+    admin_service.proto
+```
+
+## Запуск контроллера
+```bash
+cd ../controller
+bazel run //cmd/grpc_server:grpc_server
+```
+
+## Запуск клиента
+```bash
+python admin.py --file config.toml
+```

admin/admin.py

@@ -0,0 +1,42 @@
+import argparse
+import grpc
+from dotenv import load_dotenv
+import os
+
+import admin_service_pb2
+import admin_service_pb2_grpc
+
+load_dotenv(".env")
+
+class AdminClient:
+    def __init__(self):
+        self.host: str = os.environ["SERVER_HOST"]
+        self.port: str = os.environ["SERVER_PORT"]
+        self.channel: grpc.channel = grpc.insecure_channel(f"{self.host}:{self.port}")
+        self.stub: admin_service_pb2_grpc.AdminServiceStub = admin_service_pb2_grpc.AdminServiceStub(self.channel)
+
+    def load_config(self, toml_file: str) -> admin_service_pb2.LoadConfigResponse:
+        with open(toml_file, "rb") as f:
+            content: bytes = f.read()
+
+        request: admin_service_pb2.LoadConfigRequest = (admin_service_pb2.LoadConfigRequest(config_data=content))
+        return self.stub.LoadConfig(request)
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--file", default="config.toml")
+    args = parser.parse_args()
+
+    try:
+        client = AdminClient()
+        client.load_config(args.file)
+        print("Config loaded")
+    except Exception as e:
+        print(f"Error loading config: {e}")
+        return 1
+
+    return 0
+
+if __name__ == "__main__":
+    main()
+

admin/admin_service.proto

@@ -0,0 +1,17 @@
+syntax = "proto3";
+
+package admin_service;
+
+option go_package = "github.com/moevm/grpc_server/pkg/proto/admin_service";
+
+service AdminService {
+    rpc LoadConfig(LoadConfigRequest) returns (LoadConfigResponse) {}
+}
+
+message LoadConfigRequest {
+    bytes config_data = 1;
+}
+
+message LoadConfigResponse {
+    bool success = 1;
+}

admin/config.toml

@@ -0,0 +1,22 @@
+[global.rules]
+block_categories = ["MALWARE", "SOCIAL"]
+block_domains = ["youtube.com", "tiktok.com"]
+allow_domains = ["github.com", "stackoverflow.com"]
+
+[global.rules.block_by_trust]
+ENTERTAINMENT = 6
+NEWS = 4
+
+[filters.filter_1]
+block_categories = ["SOCIAL", "ENTERTAINMENT"]
+block_domains = ["instagram.com"]
+allow_domains = ["vk.com"]
+
+[filters.filter_1.block_by_trust]
+SOCIAL = 8
+ENTERTAINMENT = 7
+
+[filters.filter_2]
+block_categories = ["MALWARE"]
+allow_domains = ["github.com", "gitlab.com"]
+min_trust_level = 3

admin/requirements.txt

@@ -0,0 +1,4 @@
+grpcio==1.60.0
+grpcio-tools==1.60.0
+protobuf==4.25.1
+python-dotenv==1.0.0

controller/MODULE.bazel

@@ -25,6 +25,7 @@ use_repo(
     go_deps,
     "com_github_joho_godotenv",
     "com_github_stretchr_testify",
+    "com_github_pelletier_go_toml",
     "org_golang_google_grpc",
     "org_golang_google_protobuf",
     "org_golang_x_text",

controller/cmd/grpc_server/BUILD

@@ -15,7 +15,9 @@ go_library(
         "//internal/config",
         "//internal/grpcserver",
         "//internal/manager",
+        "//pkg/proto/admin_service:admin_service_go_proto",
         "//pkg/proto/file_service:file_service_go_proto",
+        "//pkg/proto/communication:communication_go_proto",
         "@org_golang_google_grpc//:go_default_library",
         "@org_golang_google_grpc//reflection",
     ],

controller/cmd/grpc_server/main.go

@@ -10,16 +10,22 @@ import (
 	pb "github.com/moevm/grpc_server/pkg/proto/file_service"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
+	adminPb "github.com/moevm/grpc_server/pkg/proto/admin_service"
+	commPb "github.com/moevm/grpc_server/pkg/proto/communication"
 )
 
 func main() {
 	cfg := config.Load()
-
+	adminServer := grpcserver.NewAdminServer()
 	mgr, err := manager.NewManager()
 	if err != nil {
 		log.Fatalf("manager.NewManager(): %v", err)
 	}
 
+	adminServer.SetManager(mgr)
+
+	dataServer := grpcserver.NewDataServer(mgr)
+
 	lis, err := net.Listen("tcp", net.JoinHostPort(cfg.Host, cfg.Port))
 	if err != nil {
 		log.Fatalf("failed to listen: %v", err)
@@ -31,7 +37,9 @@ func main() {
 	}
 
 	service := grpc.NewServer(serverOpts...)
+	adminPb.RegisterAdminServiceServer(service, adminServer)
 	pb.RegisterFileServiceServer(service, grpcserver.NewServer(cfg.AllowedChars, mgr))
+	commPb.RegisterDataServiceServer(service, dataServer)  
 	reflection.Register(service)
 
 	log.Printf("Server starting on %s:%s", cfg.Host, cfg.Port)

controller/go.mod

@@ -11,6 +11,7 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	golang.org/x/net v0.40.0 // indirect
 	golang.org/x/sys v0.33.0 // indirect

controller/go.sum

@@ -12,6 +12,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
 github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
+github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3ve8=
+github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=

controller/internal/grpcserver/BUILD

@@ -2,21 +2,29 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "grpcserver",
-    srcs = ["server.go"],
+    srcs = [
+        "server.go",
+        "admin_server.go",
+        "data_server.go",
+    ],
     importpath = "github.com/moevm/grpc_server/internal/grpcserver",
     visibility = ["//visibility:public"],
     deps = [
         "//internal/manager",
         "//pkg/proto/file_service:file_service_go_proto",
+        "//pkg/proto/admin_service:admin_service_go_proto",
+        "//pkg/proto/communication:communication_go_proto",
         "@org_golang_google_grpc//codes",
         "@org_golang_google_grpc//status",
-    ],
+        "@org_golang_google_grpc//:go_default_library",
+        "@org_golang_google_protobuf//types/known/emptypb", 
+    ]
 )
 
 go_test(
     name = "grpcserver_test",
     srcs = ["server_test.go"],
-    embed = [ ":grpcserver" ],
+    embed = [":grpcserver"],
     deps = [
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",

controller/internal/grpcserver/admin_server.go

@@ -0,0 +1,37 @@
+package grpcserver
+
+import (
+	"context"
+	"github.com/moevm/grpc_server/internal/manager"
+	pb "github.com/moevm/grpc_server/pkg/proto/admin_service"
+)
+
+type AdminServer struct {
+	pb.UnimplementedAdminServiceServer
+	configData []byte
+	manager    *manager.Manager 
+}
+
+func NewAdminServer() *AdminServer {
+	return &AdminServer{}
+}
+
+func (s *AdminServer) SetManager(m *manager.Manager) {
+	s.manager = m
+}
+
+func (s *AdminServer) LoadConfig(ctx context.Context, req *pb.LoadConfigRequest) (*pb.LoadConfigResponse, error) {
+	s.configData = req.ConfigData
+
+	if s.manager != nil {
+		s.manager.UpdateConfig(s.configData)
+	}
+
+	return &pb.LoadConfigResponse{
+		Success: true,
+	}, nil
+}
+
+func (s *AdminServer) GetConfig() []byte {
+	return s.configData
+}

controller/internal/grpcserver/data_server.go

@@ -0,0 +1,44 @@
+package grpcserver
+
+import (
+    "context"
+    "log"
+
+    "github.com/moevm/grpc_server/internal/manager"
+    pb "github.com/moevm/grpc_server/pkg/proto/communication"
+    "google.golang.org/grpc/codes"
+    "google.golang.org/grpc/status"
+    "google.golang.org/protobuf/types/known/emptypb"
+)
+
+type DataServer struct {
+    pb.UnimplementedDataServiceServer
+    manager *manager.Manager
+}
+
+func NewDataServer(mgr *manager.Manager) *DataServer {
+    return &DataServer{manager: mgr}
+}
+
+func (s *DataServer) GetPolicy(ctx context.Context, req *pb.GetPolicyRequest) (*pb.WorkerPolicy, error) {
+    log.Printf("gRPC GetPolicy from worker %d (hash: %d)", req.WorkerId, req.PolicyHash)
+    policy := s.manager.GetWorkerPolicy(req.WorkerId)
+    if policy == nil {
+        return nil, status.Errorf(codes.NotFound, "no policy for worker %d", req.WorkerId)
+    }
+    return policy, nil
+}
+
+func (s *DataServer) Classify(ctx context.Context, req *pb.ClassifyRequest) (*pb.ClassifyResponse, error) {
+    log.Printf("gRPC Classify from worker %d for domain: %s", req.WorkerId, req.Domain)
+    return &pb.ClassifyResponse{
+        Categories: []string{"unknown"},
+        TrustLevel: 50,
+    }, nil
+}
+
+func (s *DataServer) SendStats(ctx context.Context, report *pb.StatsReport) (*emptypb.Empty, error) {
+    log.Printf("gRPC Stats from worker %d: blocked=%d allowed=%d", 
+        report.WorkerId, report.TotalBlocked, report.TotalAllowed)
+    return &emptypb.Empty{}, nil
+}

controller/internal/manager/BUILD

@@ -2,14 +2,15 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "manager",
-    srcs = ["manager.go"],
+    srcs = ["manager.go",  "policy_manager.go",],
     importpath = "github.com/moevm/grpc_server/internal/manager",
     visibility = ["//visibility:public"],
     deps = [
         "//internal/conn",
-        "//pkg/converter",
         "//pkg/proto/communication:communication_go_proto",
         "@org_golang_google_protobuf//proto",
         "@com_github_joho_godotenv//:go_default_library",
+        "@com_github_pelletier_go_toml//:go_default_library",
+        "@org_golang_google_protobuf//types/known/structpb",
     ],
 )