One Hat Cyber Team

Your IP: 18.118.37.74
Server IP: 97.74.87.16
Server: Linux 16.87.74.97.host.secureserver.net 5.14.0-503.38.1.el9_5.x86_64 #1 SMP PREEMPT_DYNAMIC Fri Apr 18 08:52:10 EDT 2025 x86_64
Server Software: Apache
PHP Version: 8.2.28
Buat File | Buat Folder

View File Name : iperf3.py

import json
import logging
import os
import subprocess
import requests

class Iperf3Runner:
    default_start_url = 'https://speed.fortisase.com/start_iperf'
    cached_start_info = None
    def __init__(self, iperf3_bin, start_url=None):
        if start_url is None:
            start_url = self.default_start_url
        self.start_url = start_url
        self.iperf_bin = iperf3_bin
        stat = os.stat(self.iperf_bin)
        perms = oct(stat.st_mode)[-3:]
        if '744' != perms:
            os.chmod(iperf3_bin, 0o744)
        self.log = logging.getLogger('Agent')

    def _fetch_info(self):
        if self.cached_start_info is not None:
            return self.cached_start_info
        r = requests.get(
            self.start_url,
            timeout=10
        )
        if 200 != r.status_code:
            raise Exception(f'Start_perf failure: {r.status_code}')
        self.cached_start_info = r.json()
        return self.cached_start_info

    def _run_test(self, iperf_args, response_key):
        p = subprocess.run(
            iperf_args,
            shell=True,
            capture_output=True,
            check=True,
            encoding='utf-8'
        )
        response = json.loads(p.stdout)
        bps = response['end']['streams'][0][response_key]['bits_per_second']
        metric_name = 'download speed'
        if response_key == 'sender':
            metric_name = 'upload speed'
        self.log.info('Metric {} args {} -> {} bits/s'.format(
            metric_name,
            iperf_args,
            bps
        ))
        return bps
            
    def get_download_speed(self):
        info = self._fetch_info()
        args = '{} -c {} -p {} -t 10 -R --json'.format(
            self.iperf_bin,
            info['ip'],
            info['upload_port']
        )
        return self._run_test(iperf_args=args, response_key='receiver')

    def get_upload_speed(self):
        info = self._fetch_info()
        args = '{} -c {} -p {} -t 10 --json'.format(
            self.iperf_bin,
            info['ip'],
            info['download_port']
        )
        return self._run_test(iperf_args=args, response_key='sender')