1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
use crate::{
id_manager::IDManager,
router::Router,
session::Session,
types::{ConnectionID, GlobalVars},
BanManager, ConfigManager, Connection, Result, SessionList,
};
use std::sync::Arc;
use tokio::time::{sleep, Duration, Instant};
use tokio_util::sync::CancellationToken;
use tracing::{enabled, error, trace, warn, Level};
//@todo finish up the logging in this
pub(crate) struct Handler<State, CState>
where
CState: Send + Sync + Clone + 'static,
{
//No Cleanup needed
pub(crate) id: ConnectionID,
pub(crate) ban_manager: BanManager,
pub(crate) id_manager: IDManager,
pub(crate) session_list: SessionList<CState>,
pub(crate) config_manager: ConfigManager,
// Not sure, but should test
pub(crate) router: Arc<Router<State, CState>>,
pub(crate) state: State,
pub(crate) connection_state: CState,
// Cleanup needed
pub(crate) connection: Connection,
pub(crate) cancel_token: CancellationToken,
pub(crate) global_vars: GlobalVars,
}
impl<State: Clone + Send + Sync + 'static, CState: Default + Clone + Send + Sync + 'static>
Handler<State, CState>
{
pub(crate) async fn run(mut self) -> Result<()> {
let address = if self.config_manager.proxy_protocol() {
self.connection.proxy_protocol().await?
} else {
self.connection.address
};
if self.config_manager.ban_manager_enabled() {
self.ban_manager.check_banned(address)?;
}
let (mut reader, tx, handle) = self.connection.init();
let session_id = self.id_manager.allocate_session_id()?;
let session_cancel_token = self.cancel_token.child_token();
let session = Session::new(
self.id.clone(),
session_id,
address,
tx,
self.config_manager.clone(),
session_cancel_token.clone(),
self.connection_state,
)?;
trace!(
id = ?self.id,
ip = &address.to_string(),
"Connection initialized",
);
self.session_list.add_miner(address, session.clone());
let sleep = sleep(Duration::from_secs(
self.config_manager.connection_config().inital_timeout,
));
tokio::pin!(sleep);
//@todo we can return a value from this loop -> break can return a value, and so we may
//want to return an error if there is one so that we can report it at the end.
while !self.cancel_token.is_cancelled() {
if session.is_disconnected() {
trace!( id = ?self.id, ip = &address.to_string(), "Session disconnected.");
break;
}
let maybe_frame = tokio::select! {
res = reader.read_frame() => {
match res {
Err(e) => {
warn!(ip = session.ip().to_string(), "Session: {} errored with the following error: {}", session.id(), e);
break;
},
Ok(frame) => frame,
}
},
() = &mut sleep => {
if enabled!(Level::DEBUG) {
error!( id = &self.id.to_string(), ip = &address.to_string(), "Session Parse Frame Timeout");
}
break;
},
//@todo we might want timeouts to reduce difficulty as well here. -> That is
//handled in retarget, so let's check that out.
() = session_cancel_token.cancelled() => {
//@todo work on these errors,
if enabled!(Level::DEBUG) {
error!( id = &self.id.to_string(), ip = &address.to_string(), "Session Disconnected");
}
break;
},
() = self.cancel_token.cancelled() => {
// If a shutdown signal is received, return from `run`.
// This will result in the task terminating.
break;
}
};
let Some(frame) = maybe_frame else {
break;
};
//Resets the Session's last active, to detect for unactive connections
session.active();
//@todo if a miner fails a function, like subscribe / authorize we don't catch it, and
//they can spam us.
//Calls the Stratum method on the router.
self.router
.call(
frame,
self.state.clone(),
//@todo would it be possible to pass session by reference?
session.clone(),
self.global_vars.clone(),
)
.await;
//Reset sleep as later as possible
sleep.as_mut().reset(Instant::now() + session.timeout());
}
trace!(
id = &self.id.to_string(),
ip = &address.to_string(),
"Connection shutdown started",
);
self.session_list.remove_miner(address);
self.id_manager.remove_session_id(session_id);
if session.needs_ban() {
self.ban_manager.add_ban(address);
}
session.shutdown();
self.cancel_token.cancel();
//@todo we should also have a timeout here - but I may change write loop so we'll see
if let Err(e) = handle.await {
trace!(id = ?self.id, cause = ?e, "Write loop error");
}
trace!(
id = ?self.id,
ip = &address.to_string(),
"Connection shutdown complete",
);
Ok(())
}
}
//@todo I big think that I think we need to focus on today is catching attacks like open sockets
//doing nothing, socketrs trying to flood, etc.
//Let's make sure we have an entire folder of tests for "attacks" and make sure that we cover them
//thoroughly.