python – 水色（仮）

#!/usr/bin/python3
# -*- coding:utf-8 -*-

import socket, select, json

TOKEN = '1234567'
HOST = '127.0.0.1'
PORT = 50009
SOCKET_TIMEOUT = 3  # seconds
BACKLOG = 10
BUFSIZE = 4096

# See: http://memo.saitodev.com/home/python_network_programing/
def initSocket():
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    readfds = set([server_sock])
    try:
        server_sock.bind((HOST, PORT))
        server_sock.listen(BACKLOG)

        while True:
            rready, wready, xready = select.select(readfds, [], [])
            for sock in rready:
                if sock is server_sock:
                    # fd = sock.fileno()
                    # print("fd: %d" % fd)
                    sock.settimeout(SOCKET_TIMEOUT)
                    conn, address = server_sock.accept()
                    readfds.add(conn)
                else:
                    buffer = sock.recv(BUFSIZE)
                    if len(buffer) == 0:
                        sock.close()
                        readfds.remove(sock)
                    else:
                        ret = getSendValue(command(buffer))
                        sock.send(ret)
                        #
                        sock.close()
                        readfds.remove(sock)
    except Exception as e:
        print("Exception: " + str(e))
    finally:
        for sock in readfds:
            sock.close()
        server_sock.close()

# @param buffer receive buffer data
# @return dict
def command(buffer):
    try:
        data = getReceiveValue(buffer)
        print("Receive: ", data)

        if 'token' not in data or data['token'] != TOKEN:
            msg = "token error."
            print(msg)
            return {"message": msg}
        if 'command' not in data:
            msg = "Receive message is failed, key 'command' not found."
            print(msg)
            return {"message": msg}
        cmd = data['command']
        return {"Receive command": cmd}
    except Exception as e:
        return {"exception": str(e)}

def getReceiveValue(buffer):
    return json.JSONDecoder().decode(buffer.decode('utf-8').strip())

def getSendValue(value):
    if not (isinstance(value, list) or isinstance(value, dict) or isinstance(value, tuple)):
        value = {value}
    return bytes(json.JSONEncoder().encode(value), 'utf-8')

def main():
    print("Backend start")
    while True:
        initSocket()
    print("Escape main loop...")

if __name__ == '__main__':
    main()

#!/usr/bin/python3

# -*- coding:utf-8 -*-

import socket, select, json

TOKEN = '1234567'

HOST = '127.0.0.1'

PORT = 50009

SOCKET_TIMEOUT = 3 # seconds

BACKLOG = 10

BUFSIZE = 4096

# See: http://memo.saitodev.com/home/python_network_programing/

def initSocket():

server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

readfds = set([server_sock])

try:

server_sock.bind((HOST, PORT))

server_sock.listen(BACKLOG)

while True:

rready, wready, xready = select.select(readfds, [], [])

for sock in rready:

if sock is server_sock:

# fd = sock.fileno()

# print("fd: %d" % fd)

sock.settimeout(SOCKET_TIMEOUT)

conn, address = server_sock.accept()

readfds.add(conn)

else:

buffer = sock.recv(BUFSIZE)

if len(buffer) == 0:

sock.close()

readfds.remove(sock)

else:

ret = getSendValue(command(buffer))

sock.send(ret)

sock.close()

readfds.remove(sock)

except Exception as e:

print("Exception: " + str(e))

finally:

for sock in readfds:

sock.close()

server_sock.close()

# @param buffer receive buffer data

# @return dict

def command(buffer):

try:

data = getReceiveValue(buffer)

print("Receive: ", data)

if 'token' not in data or data['token'] != TOKEN:

msg = "token error."

print(msg)

return {"message": msg}

if 'command' not in data:

msg = "Receive message is failed, key 'command' not found."

print(msg)

return {"message": msg}

cmd = data['command']

return {"Receive command": cmd}

except Exception as e:

return {"exception": str(e)}

def getReceiveValue(buffer):

return json.JSONDecoder().decode(buffer.decode('utf-8').strip())

def getSendValue(value):

if not (isinstance(value, list) or isinstance(value, dict) or isinstance(value, tuple)):

value = {value}

return bytes(json.JSONEncoder().encode(value), 'utf-8')

def main():

print("Backend start")

while True:

initSocket()

print("Escape main loop...")

if __name__ == '__main__':

main()

<?php

$sendMessage = array(
    'cmd' => 'commandA',
);
$result = commBackend($sendMessage);
print_r($result);

function commBackend($sendMessage) {

    $sendMessage['token'] = '1234567';

    $address = '127.0.0.1';
    $service_port = 50009;
    $timeout_sec = 3;   // sec
    $timeout = time() + $timeout_sec;

    try {
        $fp = fsockopen($address, $service_port);
        if (!$fp) {
            throw new Exception('Cannot communicate Backend.');
        }
        // non blocking, need to timeout.
        stream_set_blocking($fp ,0);
        stream_set_timeout($fp, $timeout_sec);

        fwrite($fp, json_encode($sendMessage));

        $contents = '';
        do {
            if ($timeout - time() <= 0) {
                $timeout = -1;
                break;
            }
            // echo '.';
            $contents .= fread($fp, 8192);
        } while (!feof($fp));

        $info = stream_get_meta_data($fp);
        // print_r($info);
        fclose($fp);

        if ($info['timed_out'] || $timeout === -1) {
            throw new Exception('Communication with the backend timed out.');
        }

        // echo $contents;
        $res = json_decode($contents);    // json string to object
        // print_r($res);
        return $res;

    } catch (Exception $e) {
        echo $e->getMessage();
        throw new Exception($e->getMessage());
    }
}

<?php

$sendMessage = array(

'cmd' => 'commandA',

);

$result = commBackend($sendMessage);

print_r($result);

function commBackend($sendMessage) {

$sendMessage['token'] = '1234567';

$address = '127.0.0.1';

$service_port = 50009;

$timeout_sec = 3; // sec

$timeout = time() + $timeout_sec;

try {

$fp = fsockopen($address, $service_port);

if (!$fp) {

throw new Exception('Cannot communicate Backend.');

}

// non blocking, need to timeout.

stream_set_blocking($fp ,0);

stream_set_timeout($fp, $timeout_sec);

fwrite($fp, json_encode($sendMessage));

$contents = '';

do {

if ($timeout - time() <= 0) {

$timeout = -1;

break;

}

// echo '.';

$contents .= fread($fp, 8192);

} while (!feof($fp));

$info = stream_get_meta_data($fp);

// print_r($info);

fclose($fp);

if ($info['timed_out'] || $timeout === -1) {

throw new Exception('Communication with the backend timed out.');

}

// echo $contents;

$res = json_decode($contents); // json string to object

// print_r($res);

return $res;

} catch (Exception $e) {

echo $e->getMessage();

throw new Exception($e->getMessage());

}

#!/usr/bin/python3
# -*- coding: utf-8 -*-

"""
Use subprocess and pamtester to check if username and password match.
pamtester needs to be executed with administrator authority.

subprocess と pamtester を使用してユーザー名とパスワードが合致するか調べる
pamtester は管理者権限で実行する必要が有る
"""

import os, pwd
import subprocess

# See: https://qiita.com/quenhulu/items/61edc5dffa711b08d64c
def getuser():
    """Get the username from the environment or password database.

    First try various environment variables, then the password
    database.  This works on Windows as long as USERNAME is set.

    """
    for name in ('LOGNAME', 'USER', 'LNAME', 'USERNAME'):
        user = os.environ.get(name)
        if user:
            return user

    # If this fails, the exception will "explain" why
    return pwd.getpwuid(os.getuid())[0]


# @return: dictionary
def execCmd(cmd, outText = ""):
    commandName = cmd.split(" ", 2)[0]
    ps = subprocess.Popen("/usr/bin/which '%s'" % commandName, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    stdout, stderr = ps.communicate()
    if ps.returncode != 0:
        raise Exception("execCmd: Command not found on PATH: %s" % commandName)

    ps = subprocess.Popen(cmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    if outText == "":
        stdout, stderr = ps.communicate()
    else:
        stdout, stderr = ps.communicate(input=outText.encode())
    return {"returncode": ps.returncode, "stdout": stdout.decode('utf-8').strip(), "stderr": stderr.decode('utf-8').strip()}

def isMatchedUsernameAndPass(username, password):
    try:
        # ret = execCmd("sudo pamtester auth %s authenticate" % username, password)
        ret = execCmd("pamtester auth %s authenticate" % username, password)
        if ret["returncode"] == 0:
            print("Success")
            return True
        else:
            print("Failure")
            return False
        return ret
    except:
        print("Error")
        return False

if __name__ == '__main__':
    you = getuser()
    if you != 'root':
        print("Needs to be executed with administrator authority. you are %s." % you)
        exit(1)
    ret = isMatchedUsernameAndPass("USERNAME", "PASSWORD")
    print(ret)

#!/usr/bin/python3

# -*- coding: utf-8 -*-

"""

Use subprocess and pamtester to check if username and password match.

pamtester needs to be executed with administrator authority.

subprocess と pamtester を使用してユーザー名とパスワードが合致するか調べる

pamtester は管理者権限で実行する必要が有る

"""

import os, pwd

import subprocess

# See: https://qiita.com/quenhulu/items/61edc5dffa711b08d64c

def getuser():

"""Get the username from the environment or password database.

First try various environment variables, then the password

database. This works on Windows as long as USERNAME is set.

"""

for name in ('LOGNAME', 'USER', 'LNAME', 'USERNAME'):

user = os.environ.get(name)

if user:

return user

# If this fails, the exception will "explain" why

return pwd.getpwuid(os.getuid())[0]

# @return: dictionary

def execCmd(cmd, outText = ""):

commandName = cmd.split(" ", 2)[0]

ps = subprocess.Popen("/usr/bin/which '%s'" % commandName, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)

stdout, stderr = ps.communicate()

if ps.returncode != 0:

raise Exception("execCmd: Command not found on PATH: %s" % commandName)

ps = subprocess.Popen(cmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)

if outText == "":

stdout, stderr = ps.communicate()

else:

stdout, stderr = ps.communicate(input=outText.encode())

return {"returncode": ps.returncode, "stdout": stdout.decode('utf-8').strip(), "stderr": stderr.decode('utf-8').strip()}

def isMatchedUsernameAndPass(username, password):

try:

# ret = execCmd("sudo pamtester auth %s authenticate" % username, password)

ret = execCmd("pamtester auth %s authenticate" % username, password)

if ret["returncode"] == 0:

print("Success")

return True

else:

print("Failure")

return False

return ret

except:

print("Error")

return False

if __name__ == '__main__':

you = getuser()

if you != 'root':

print("Needs to be executed with administrator authority. you are %s." % you)

exit(1)

ret = isMatchedUsernameAndPass("USERNAME", "PASSWORD")

print(ret)

タグ: python

Socket sync communication test between PHP7 and Python3

Use subprocess and pamtester to check if username and password match. / subprocess と pamtester を使用してユーザー名とパスワードが合致するか調べるメモ